Merge branch 'main' of github.com:RolnickLab/antenna into deployments/ood.antenna.insectai.org

Author: mihow

.envs/.ci/.django

@@ -16,3 +16,6 @@ MINIO_DEFAULT_BUCKET=ami-ci
 MINIO_STORAGE_USE_HTTPS=False
 MINIO_TEST_BUCKET=ami-test-ci
 MINIO_BROWSER_REDIRECT_URL=http://minio:9001
+
+DEFAULT_PROCESSING_SERVICE_NAME=Test Processing Service
+DEFAULT_PROCESSING_SERVICE_ENDPOINT=http://ml_backend:2000

.envs/.local/.django

@@ -37,3 +37,8 @@ MINIO_DEFAULT_BUCKET=ami
 MINIO_STORAGE_USE_HTTPS=False
 MINIO_TEST_BUCKET=ami-test
 MINIO_BROWSER_REDIRECT_URL=http://minio:9001
+
+# Default processing service (local)
+DEFAULT_PROCESSING_SERVICE_NAME=Local Processing Service
+DEFAULT_PROCESSING_SERVICE_ENDPOINT=http://ml_backend:2000
+# DEFAULT_PIPELINES_ENABLED=random,constant  # When set to None, all pipelines will be enabled.

.envs/.production/.django-example

@@ -53,3 +53,8 @@ DJANGO_ACCOUNT_ALLOW_REGISTRATION=True
 # Gunicorn
 # ------------------------------------------------------------------------------
 WEB_CONCURRENCY=4
+
+# Default processing service
+DEFAULT_PROCESSING_SERVICE_NAME="AMI Data Companion"
+DEFAULT_PROCESSING_SERVICE_ENDPOINT=https://ml.antenna.insectai.org/
+DEFAULT_PIPELINES_ENABLED=global_moths_2024,quebec_vermont_moths_2023,panama_moths_2023,uk_denmark_moths_2023

ami/base/models.py

@@ -1,4 +1,6 @@
+from django.contrib.auth.models import AbstractUser, AnonymousUser
 from django.db import models
+from guardian.shortcuts import get_perms
 
 import ami.tasks
 
@@ -29,5 +31,94 @@ def update_calculated_fields(self, *args, **kwargs):
         """Update calculated fields specific to each model."""
         pass
 
+    def _get_object_perms(self, user):
+        """
+        Get the object-level permissions for the user on this instance.
+        This method retrieves permissions like `update_modelname`, `create_modelname`, etc.
+        """
+        project = self.get_project()
+        if not project:
+            return []
+
+        model_name = self._meta.model_name
+        all_perms = get_perms(user, project)
+        object_perms = [perm for perm in all_perms if perm.endswith(f"_{model_name}")]
+        return object_perms
+
+    def check_permission(self, user: AbstractUser | AnonymousUser, action: str) -> bool:
+        """
+        Check if the user has permission to perform the action
+        on this instance.
+        This method is used to determine if the user can perform
+        CRUD operations or custom actions on the model instance.
+        """
+        project = self.get_project() if hasattr(self, "get_project") else None
+        if not project:
+            return False
+        if action == "retrieve":
+            # Allow view
+            return True
+
+        model = self._meta.model_name
+        crud_map = {
+            "create": f"create_{model}",
+            "update": f"update_{model}",
+            "partial_update": f"update_{model}",
+            "destroy": f"delete_{model}",
+        }
+
+        if action in crud_map:
+            return user.has_perm(crud_map[action], project)
+
+        # Delegate to model-specific logic
+        return self.check_custom_permission(user, action)
+
+    def check_custom_permission(self, user: AbstractUser | AnonymousUser, action: str) -> bool:
+        """Check custom permissions for the user on this instance.
+        This is used for actions that are not standard CRUD operations.
+        """
+        assert self._meta.model_name is not None, "Model must have a model_name defined in Meta class."
+        model_name = self._meta.model_name.lower()
+        permission_codename = f"{action}_{model_name}"
+        project = self.get_project() if hasattr(self, "get_project") else None
+
+        return user.has_perm(permission_codename, project)
+
+    def get_user_object_permissions(self, user) -> list[str]:
+        """
+        Returns a list of object-level permissions the user has on this instance.
+        This is used by frontend to determine what actions the user can perform.
+        """
+        # Return all permissions for superusers
+        if user.is_superuser:
+            allowed_custom_actions = self.get_custom_user_permissions(user)
+            return ["update", "delete"] + allowed_custom_actions
+
+        object_perms = self._get_object_perms(user)
+        # Check for update and delete permissions
+        allowed_actions = set()
+        for perm in object_perms:
+            action = perm.split("_", 1)[0]
+            if action in {"update", "delete"}:
+                allowed_actions.add(action)
+
+        allowed_custom_actions = self.get_custom_user_permissions(user)
+        allowed_actions.update(set(allowed_custom_actions))
+        return list(allowed_actions)
+
+    def get_custom_user_permissions(self, user: AbstractUser | AnonymousUser) -> list[str]:
+        """
+        Returns a list of custom permissions (not standard CRUD actions) that the user has on this instance.
+        """
+        object_perms = self._get_object_perms(user)
+        custom_perms = set()
+        # Extract custom permissions that are not standard CRUD actions
+        for perm in object_perms:
+            action = perm.split("_", 1)[0]
+            # Make sure to exclude standard CRUD actions
+            if action not in ["view", "create", "update", "delete"]:
+                custom_perms.add(action)
+        return list(custom_perms)
+
     class Meta:
         abstract = True

ami/base/permissions.py

@@ -6,19 +6,7 @@
 from guardian.shortcuts import get_perms
 from rest_framework import permissions
 
-from ami.jobs.models import Job
-from ami.main.models import (
-    BaseModel,
-    Deployment,
-    Device,
-    Project,
-    S3StorageSource,
-    Site,
-    SourceImage,
-    SourceImageCollection,
-    SourceImageUpload,
-)
-from ami.users.roles import ProjectManager
+from ami.main.models import BaseModel
 
 logger = logging.getLogger(__name__)
 
@@ -64,18 +52,8 @@ def add_object_level_permissions(
     """
 
     permissions = response_data.get("user_permissions", set())
-    project = instance.get_project() if hasattr(instance, "get_project") else None
-    model_name = instance._meta.model_name  # Get model name
-    if user and user.is_superuser:
-        permissions.update(["update", "delete"])
-
-    if project:
-        user_permissions = get_perms(user, project)
-        # Filter and extract only the action part of "action_modelname" based on instance type
-        filtered_permissions = filter_permissions(permissions=user_permissions, model_name=model_name)
-        # Do not return create, view permissions at object-level
-        filtered_permissions -= {"create", "view"}
-        permissions.update(filtered_permissions)
+    if isinstance(instance, BaseModel):
+        permissions.update(instance.get_user_object_permissions(user))
     response_data["user_permissions"] = list(permissions)
     return response_data
 
@@ -99,165 +77,13 @@ def add_collection_level_permissions(user: User | None, response_data: dict, mod
     return response_data
 
 
-class CRUDPermission(permissions.BasePermission):
+class ObjectPermission(permissions.BasePermission):
     """
-    Generic CRUD permission class that dynamically checks user permissions on an object.
-    Permission names follow the convention: `create_<model>`, `update_<model>`, `delete_<model>`.
+    Generic permission class that delegates to the model's `check_permission(user, action)` method.
     """
 
-    model = None
-
     def has_permission(self, request, view):
-        """Handles general permission checks"""
-
-        return True  # Fallback to object level permissions
-
-    def has_object_permission(self, request, view, obj):
-        """Handles object-level permission checks."""
-        model_name = self.model._meta.model_name
-        project = obj.get_project() if hasattr(obj, "get_project") else None
-        # check for create action
-        if view.action == "create":
-            return request.user.is_superuser or request.user.has_perm(f"create_{model_name}", project)
-
-        if view.action == "retrieve":
-            return True  # Allow all users to view objects
-
-        # Map ViewSet actions to permission names
-        action_perms = {
-            "retrieve": f"view_{model_name}",
-            "update": f"update_{model_name}",
-            "partial_update": f"update_{model_name}",
-            "destroy": f"delete_{model_name}",
-        }
-
-        required_perm = action_perms.get(view.action)
-        if not required_perm:
-            return True
-        return request.user.has_perm(required_perm, project)
-
-
-class ProjectCRUDPermission(CRUDPermission):
-    model = Project
-
-
-class JobCRUDPermission(CRUDPermission):
-    model = Job
-
-
-class DeploymentCRUDPermission(CRUDPermission):
-    model = Deployment
-
-
-class SourceImageCollectionCRUDPermission(CRUDPermission):
-    model = SourceImageCollection
-
-
-class SourceImageUploadCRUDPermission(CRUDPermission):
-    model = SourceImageUpload
-
-
-class SourceImageCRUDPermission(CRUDPermission):
-    model = SourceImage
-
-
-class CanStarSourceImage(permissions.BasePermission):
-    """Custom permission to check if the user can star a Source image."""
-
-    permission = Project.Permissions.STAR_SOURCE_IMAGE
-
-    def has_object_permission(self, request, view, obj):
-        if view.action in ["unstar", "star"]:
-            project = obj.get_project() if hasattr(obj, "get_project") else None
-            return request.user.has_perm(self.permission, project)
-        return True
-
-
-class S3StorageSourceCRUDPermission(CRUDPermission):
-    model = S3StorageSource
-
-
-class SiteCRUDPermission(CRUDPermission):
-    model = Site
-
-
-class DeviceCRUDPermission(CRUDPermission):
-    model = Device
-
-
-# Identification permission checks
-class CanUpdateIdentification(permissions.BasePermission):
-    """Custom permission to check if the user can update/create an identification."""
-
-    permission = Project.Permissions.UPDATE_IDENTIFICATION
-
-    def has_object_permission(self, request, view, obj):
-        if view.action in ["create", "update", "partial_update"]:
-            project = obj.get_project() if hasattr(obj, "get_project") else None
-            return request.user.has_perm(self.permission, project)
-        return True
-
-
-class CanDeleteIdentification(permissions.BasePermission):
-    """Custom permission to check if the user can delete an identification."""
-
-    permission = Project.Permissions.DELETE_IDENTIFICATION
-
-    def has_object_permission(self, request, view, obj):
-        project = obj.get_project() if hasattr(obj, "get_project") else None
-        # Check if user is superuser or staff or project manager
-        if view.action == "destroy":
-            if request.user.is_superuser or ProjectManager.has_role(request.user, project):
-                return True
-            # Check if the user is the owner of the object
-            return obj.user == request.user
-        return True
-
-
-# Job run permission check
-class CanRunJob(permissions.BasePermission):
-    """Custom permission to check if the user can run a job."""
-
-    permission = Project.Permissions.RUN_JOB
-
-    def has_object_permission(self, request, view, obj):
-        if view.action == "run":
-            project = obj.get_project() if hasattr(obj, "get_project") else None
-            return request.user.has_perm(self.permission, project)
-        return True
-
-
-class CanRetryJob(permissions.BasePermission):
-    """Custom permission to check if the user can retry a job."""
-
-    permission = Project.Permissions.RETRY_JOB
-
-    def has_object_permission(self, request, view, obj):
-        if view.action == "retry":
-            project = obj.get_project() if hasattr(obj, "get_project") else None
-            return request.user.has_perm(self.permission, project)
-        return True
-
-
-class CanCancelJob(permissions.BasePermission):
-    """Custom permission to check if the user can cancel a job."""
-
-    permission = Project.Permissions.CANCEL_JOB
-
-    def has_object_permission(self, request, view, obj):
-        if view.action == "cancel":
-            project = obj.get_project() if hasattr(obj, "get_project") else None
-            return request.user.has_perm(self.permission, project)
-        return True
-
-
-class CanPopulateSourceImageCollection(permissions.BasePermission):
-    """Custom permission to check if the user can populate a collection."""
-
-    permission = Project.Permissions.POPULATE_COLLECTION
+        return True  # Always allow — object-level handles actual checks
 
-    def has_object_permission(self, request, view, obj):
-        if view.action == "populate":
-            project = obj.get_project() if hasattr(obj, "get_project") else None
-            return request.user.has_perm(self.permission, project)
-        return True
+    def has_object_permission(self, request, view, obj: BaseModel):
+        return obj.check_permission(request.user, view.action)

ami/base/views.py

@@ -22,17 +22,23 @@ class ProjectMixin:
     def get_active_project(self) -> Project | None:
         from ami.base.serializers import SingleParamSerializer
 
+        param = "project_id"
+
         project_id = None
         # Extract from URL `/projects/` is in the url path
         if "/projects/" in self.request.path:
             project_id = self.kwargs.get("pk")
 
         # If not in URL, try query parameters
         if not project_id:
+            # Look for project_id in GET query parameters or POST data
+            # POST data returns a list of ints, but QueryDict.get() returns a single value
+            project_id = self.request.query_params.get(param) or self.request.data.get(param)
+
             project_id = SingleParamSerializer[int].clean(
-                param_name="project_id",
+                param_name=param,
                 field=serializers.IntegerField(required=self.require_project, min_value=0),
-                data=self.request.query_params,
+                data={param: project_id} if project_id else {},
             )
 
         return get_object_or_404(Project, id=project_id) if project_id else None