Configurable taxa lists

ami/base/permissions.py

@@ -77,6 +77,73 @@ def add_collection_level_permissions(user: User | None, response_data: dict, mod
     return response_data
 
 
+def add_m2m_object_permissions(user, instance, project, response_data: dict) -> dict:
+    """
+    Add object-level permissions for models with an M2M relationship to Project.
+
+    The default permission resolution (BaseModel._get_object_perms) relies on
+    get_project(), which returns None for M2M-to-Project models (TaxaList, etc.)
+    because there's no single owning project. This function resolves permissions
+    against a specific project from the request context instead.
+
+    Validates that the instance actually belongs to the given project before
+    granting any permissions (prevents cross-project permission leaks).
+
+    This is a temporary approach for the M2M permission gap described in #1120.
+    Once that issue is resolved, this should be replaced by a generic permission
+    class (Pattern B: Bare M2M) that handles TaxaList, Taxon, ProcessingService,
+    Pipeline, and other M2M-to-Project models uniformly.
+    """
+    perms = set(response_data.get("user_permissions", []))
+
+    if not project or not instance.projects.filter(pk=project.pk).exists():
+        response_data["user_permissions"] = list(perms)
+        return response_data
+
+    if user.is_superuser:
+        perms.update(["update", "delete"])
+    else:
+        model_name = instance._meta.model_name
+        all_perms = get_perms(user, project)
+        for perm in all_perms:
+            if perm.endswith(f"_{model_name}"):
+                action = perm.split("_", 1)[0]
+                if action in {"update", "delete"}:
+                    perms.add(action)
+
+    response_data["user_permissions"] = list(perms)
+    return response_data
+
+
+class IsProjectMemberOrReadOnly(permissions.BasePermission):
+    """
+    Safe methods are allowed for everyone.
+    Unsafe methods (POST, PUT, PATCH, DELETE) require the requesting user to be
+    a member of the active project (resolved via ProjectMixin.get_active_project).
+    """
+
+    def has_permission(self, request, view):
+        if request.method in permissions.SAFE_METHODS:
+            return True
+
+        if not request.user or not request.user.is_authenticated:
+            return False
+
+        if request.user.is_superuser:  # type: ignore[union-attr]
+            return True
+
+        # view must provide get_active_project (i.e. use ProjectMixin)
+        get_active_project = getattr(view, "get_active_project", None)
+        if not get_active_project:
+            return False
+
+        project = get_active_project()
+        if not project:
+            return False
+
+        return project.members.filter(pk=request.user.pk).exists()
+
+
 class ObjectPermission(permissions.BasePermission):
     """
     Generic permission class that delegates to the model's `check_permission(user, action)` method.

ami/main/api/serializers.py

@@ -6,6 +6,7 @@
 from rest_framework.request import Request
 
 from ami.base.fields import DateStringField
+from ami.base.permissions import add_m2m_object_permissions
 from ami.base.serializers import DefaultSerializer, MinimalNestedModelSerializer, reverse_with_params
 from ami.base.views import get_active_project
 from ami.jobs.models import Job
@@ -633,13 +634,23 @@ def get_occurrences(self, obj):
         )
 
 
-class TaxaListSerializer(serializers.ModelSerializer):
+class TaxaListSerializer(DefaultSerializer):
     taxa = serializers.SerializerMethodField()
-    projects = serializers.PrimaryKeyRelatedField(queryset=Project.objects.all(), many=True)
+    taxa_count = serializers.SerializerMethodField()
+    projects = serializers.SerializerMethodField()
 
     class Meta:
         model = TaxaList
-        fields = ["id", "name", "description", "taxa", "projects"]
+        fields = [
+            "id",
+            "name",
+            "description",
+            "taxa",
+            "taxa_count",
+            "projects",
+            "created_at",
+            "updated_at",
+        ]
 
     def get_taxa(self, obj):
         """
@@ -651,6 +662,43 @@ def get_taxa(self, obj):
             params={"taxa_list_id": obj.pk},
         )
 
+    def get_taxa_count(self, obj):
+        """
+        Return the number of taxa in this list.
+        Uses annotated_taxa_count if available (from ViewSet) for performance.
+        """
+        return getattr(obj, "annotated_taxa_count", obj.taxa.count())
+
+    def get_permissions(self, instance, instance_data):
+        request = self.context["request"]
+        project = get_active_project(request=request)
+        return add_m2m_object_permissions(request.user, instance, project, instance_data)
+
+    def get_projects(self, obj):
+        """
+        Return list of project IDs this taxa list belongs to.
+        This is read-only and managed by the server.
+        """
+        return list(obj.projects.values_list("id", flat=True))
+
+
+class TaxaListTaxonInputSerializer(serializers.Serializer):
+    """Serializer for adding a taxon to a taxa list."""
+
+    taxon_id = serializers.IntegerField(required=True)
+
+    def validate_taxon_id(self, value):
+        """Validate that the taxon exists."""
+        if not Taxon.objects.filter(id=value).exists():
+            raise serializers.ValidationError("Taxon does not exist.")
+        return value
+
+
+class TaxaListTaxonSerializer(TaxonNoParentNestedSerializer):
+    """Serializer for taxa in a taxa list (simplified taxon representation)."""
+
+    pass
+
 
 class CaptureTaxonSerializer(DefaultSerializer):
     parent = TaxonNoParentNestedSerializer(read_only=True)

ami/main/api/views.py

@@ -9,6 +9,7 @@
 from django.db.models.functions import Coalesce
 from django.db.models.query import QuerySet
 from django.forms import BooleanField, CharField, IntegerField
+from django.shortcuts import get_object_or_404
 from django.utils import timezone
 from django_filters.rest_framework import DjangoFilterBackend
 from drf_spectacular.types import OpenApiTypes
@@ -26,7 +27,7 @@
 from ami.base.filters import NullsLastOrderingFilter, ThresholdFilter
 from ami.base.models import BaseQuerySet
 from ami.base.pagination import LimitOffsetPaginationWithPermissions
-from ami.base.permissions import IsActiveStaffOrReadOnly, ObjectPermission
+from ami.base.permissions import IsActiveStaffOrReadOnly, IsProjectMemberOrReadOnly, ObjectPermission
 from ami.base.serializers import FilterParamsSerializer, SingleParamSerializer
 from ami.base.views import ProjectMixin
 from ami.main.api.schemas import project_id_doc_param
@@ -83,6 +84,8 @@
     StorageSourceSerializer,
     StorageStatusSerializer,
     TaxaListSerializer,
+    TaxaListTaxonInputSerializer,
+    TaxaListTaxonSerializer,
     TaxonListSerializer,
     TaxonSearchResultSerializer,
     TaxonSerializer,
@@ -1261,11 +1264,15 @@ def list(self, request, *args, **kwargs):
 
 class TaxonTaxaListFilter(filters.BaseFilterBackend):
     """
-    Filters taxa based on a TaxaList Similar to `OccurrenceTaxaListFilter`.
+    Filters taxa based on a TaxaList.
 
-    Queries for all taxa that are either:
-    - Directly in the requested TaxaList.
-    - A descendant (child or deeper) of any taxon in the TaxaList, recursively.
+    By default, queries for taxa that are directly in the TaxaList and their descendants.
+    If include_descendants=false, only taxa directly in the TaxaList are returned.
+
+    Query parameters:
+    - taxa_list_id: ID of the taxa list to filter by
+    - include_descendants: Set to 'false' to exclude descendants (default: true)
+    - not_taxa_list_id: ID of taxa list to exclude
     """
 
     query_param = "taxa_list_id"
@@ -1277,11 +1284,20 @@ def filter_queryset(self, request, queryset, view):
             request.query_params.get(self.query_param_exclusive)
         )
 
+        include_descendants_default = True
+        include_descendants = request.query_params.get("include_descendants", include_descendants_default)
+        if include_descendants is not None:
+            include_descendants = BooleanField(required=False).clean(include_descendants)
+
         def _get_filter(taxa_list: TaxaList) -> models.Q:
             taxa = taxa_list.taxa.all()  # Get taxa in the taxa list
             query_filter = Q(id__in=taxa)
-            for taxon in taxa:
-                query_filter |= Q(parents_json__contains=[{"id": taxon.pk}])
+
+            # Only include descendants if explicitly requested
+            if include_descendants:
+                for taxon in taxa:
+                    query_filter |= Q(parents_json__contains=[{"id": taxon.pk}])
+
             return query_filter
 
         if taxalist_id:
@@ -1608,17 +1624,107 @@ def list(self, request, *args, **kwargs):
         return super().list(request, *args, **kwargs)
 
 
-class TaxaListViewSet(viewsets.ModelViewSet, ProjectMixin):
+class TaxaListViewSet(DefaultViewSet, ProjectMixin):
     queryset = TaxaList.objects.all()
+    serializer_class = TaxaListSerializer
+    ordering_fields = [
+        "name",
+        "description",
+        "annotated_taxa_count",
+        "created_at",
+        "updated_at",
+    ]
+    permission_classes = [IsProjectMemberOrReadOnly]
+    require_project = True
 
     def get_queryset(self):
         qs = super().get_queryset()
+        # Annotate with taxa count for better performance
+        qs = qs.annotate(annotated_taxa_count=models.Count("taxa"))
         project = self.get_active_project()
         if project:
             return qs.filter(projects=project)
         return qs
 
-    serializer_class = TaxaListSerializer
+    def perform_create(self, serializer):
+        """
+        Create a TaxaList and automatically assign it to the active project.
+
+        Users cannot manually assign taxa lists to projects for security reasons.
+        A taxa list is always created in the context of the active project.
+        """
+        instance = serializer.save()
+        project = self.get_active_project()
+        if project:
+            instance.projects.add(project)
+
+
+class TaxaListTaxonViewSet(viewsets.GenericViewSet, ProjectMixin):
+    """
+    Nested ViewSet for managing taxa in a taxa list.
+    Accessed via /taxa/lists/{taxa_list_id}/taxa/
+
+    Only provides create (POST) and delete (DELETE) actions.
+    The UI lists taxa via the main /taxa/ endpoint with a taxa_list_id filter.
+    """
+
+    serializer_class = TaxaListTaxonSerializer
+    permission_classes = [IsProjectMemberOrReadOnly]
+    require_project = True
+
+    def get_taxa_list(self):
+        """Get the parent taxa list from URL parameters, scoped to the active project."""
+        taxa_list_id = self.kwargs.get("taxalist_pk")
+        project = self.get_active_project()
+        try:
+            return TaxaList.objects.get(pk=taxa_list_id, projects=project)
+        except TaxaList.DoesNotExist:
+            raise api_exceptions.NotFound("Taxa list not found.") from None
+
+    def get_queryset(self):
+        """Return taxa in the specified taxa list."""
+        taxa_list = self.get_taxa_list()
+        return taxa_list.taxa.all()
+
+    def create(self, request, taxalist_pk=None):
+        """Add a taxon to the taxa list."""
+        taxa_list = self.get_taxa_list()
+
+        # Validate input
+        input_serializer = TaxaListTaxonInputSerializer(data=request.data)
+        input_serializer.is_valid(raise_exception=True)
+        taxon_id = input_serializer.validated_data["taxon_id"]
+
+        # Check if already exists
+        if taxa_list.taxa.filter(pk=taxon_id).exists():
+            return Response(
+                {"non_field_errors": ["Taxon is already in this taxa list."]},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        # Add taxon
+        taxon = get_object_or_404(Taxon, pk=taxon_id)
+        taxa_list.taxa.add(taxon)
+
+        # Return the added taxon
+        serializer = self.get_serializer(taxon)
+        return Response(serializer.data, status=status.HTTP_201_CREATED)
+
+    @action(detail=False, methods=["delete"], url_path=r"(?P<taxon_id>\d+)")
+    def delete_by_taxon(self, request, taxalist_pk=None, taxon_id=None):
+        """
+        Remove a taxon from the taxa list by taxon ID.
+        DELETE /taxa/lists/{taxa_list_id}/taxa/{taxon_id}/
+        """
+        taxa_list = self.get_taxa_list()
+
+        # Check if taxon exists in list
+        if not taxa_list.taxa.filter(pk=taxon_id).exists():
+            raise api_exceptions.NotFound("Taxon is not in this taxa list.")
+
+        # Remove taxon
+        taxa_list.taxa.remove(taxon_id)
+        return Response(status=status.HTTP_204_NO_CONTENT)
 
 
 class TagViewSet(DefaultViewSet, ProjectMixin):