fix(occurrences): remove all n+1 queries from occurrence API

ami/main/api/serializers.py

@@ -1318,6 +1318,7 @@ class OccurrenceListSerializer(DefaultSerializer):
     deployment = DeploymentNestedSerializer(read_only=True)
     event = EventNestedSerializer(read_only=True)
     # first_appearance = TaxonSourceImageNestedSerializer(read_only=True)
+    detection_images = serializers.SerializerMethodField()
     determination_details = serializers.SerializerMethodField()
     best_machine_prediction = serializers.SerializerMethodField()
     identifications = OccurrenceIdentificationSerializer(many=True, read_only=True)
@@ -1364,27 +1365,59 @@ class Meta:
             "updated_at",
         ]
 
-    def get_determination_details(self, obj: Occurrence):
-        # @TODO convert this to query methods to avoid N+1 queries.
-        # Currently at 100+ queries per page of 10 occurrences.
-        # Add a reusable method to the OccurrenceQuerySet class and call it from the ViewSet.
+    def get_detection_images(self, obj: Occurrence) -> list[str]:
+        """Return media URLs for the occurrence's detection crops.
 
-        context = self.context
+        Reads from the `prefetched_detection_images` to_attr list populated by
+        `prefetch_detection_images()`; falls back to the model method for
+        callers that did not apply the prefetch (e.g. detail view).
+        """
+        from ami.main.models_future.occurrence import detection_image_urls_from_prefetch
+
+        if hasattr(obj, "prefetched_detection_images"):
+            return detection_image_urls_from_prefetch(obj)
+        return list(obj.detection_images())
+
+    def _best_identification(self, obj: Occurrence) -> Identification | None:
+        """Pick the best identification, preferring prefetched data when available."""
+        from ami.main.models_future.occurrence import best_identification_from_prefetch
+
+        # `with_identifications()` populates the relation cache; if it ran we
+        # can pick the best in Python without a fresh DB hit.
+        if "identifications" in getattr(obj, "_prefetched_objects_cache", {}):
+            return best_identification_from_prefetch(obj)
+        return obj.best_identification
 
+    def _best_prediction(self, obj: Occurrence):
+        """Pick the best machine prediction, preferring prefetched data when available."""
+        from ami.main.models_future.occurrence import best_prediction_from_prefetch
+
+        # `prefetch_classifications_for_best_prediction()` populates both the
+        # detections cache and the nested classifications cache.
+        prefetch_cache = getattr(obj, "_prefetched_objects_cache", {})
+        if "detections" in prefetch_cache:
+            return best_prediction_from_prefetch(obj)
+        return obj.best_prediction
+
+    def get_determination_details(self, obj: Occurrence):
+        context = self.context
         # Add this occurrence to the context so that the nested serializers can access it
         # the `parent` attribute is not available since we are manually instantiating the serializers
         context["occurrence"] = obj
 
         taxon = TaxonNestedSerializer(obj.determination, context=context).data if obj.determination else None
-        if obj.best_identification:
-            identification = OccurrenceIdentificationSerializer(obj.best_identification, context=context).data
+
+        best_ident = self._best_identification(obj)
+        if best_ident:
+            identification = OccurrenceIdentificationSerializer(best_ident, context=context).data
         else:
             identification = None
 
-        if identification or not obj.best_prediction:
+        if identification:
             prediction = None
         else:
-            prediction = ClassificationNestedSerializer(obj.best_prediction, context=context).data
+            best_pred = self._best_prediction(obj)
+            prediction = ClassificationNestedSerializer(best_pred, context=context).data if best_pred else None
 
         return dict(
             taxon=taxon,
@@ -1402,7 +1435,7 @@ def get_best_machine_prediction(self, obj: Occurrence) -> dict | None:
         context = self.context
         context["occurrence"] = obj
 
-        prediction = obj.best_prediction
+        prediction = self._best_prediction(obj)
         if not prediction:
             return None
 

ami/main/api/views.py

@@ -1227,7 +1227,9 @@ def get_queryset(self) -> QuerySet["Occurrence"]:
         qs = qs.with_detections_count().with_timestamps()  # type: ignore
         qs = qs.with_identifications()  # type: ignore
         qs = qs.apply_default_filters(project, self.request)  # type: ignore
-        if self.action != "list":
+        if self.action == "list":
+            qs = qs.with_list_prefetches()  # type: ignore
+        else:
             qs = qs.prefetch_related(
                 Prefetch(
                     "detections", queryset=Detection.objects.order_by("-timestamp").select_related("source_image")

ami/main/models.py

@@ -2932,6 +2932,15 @@ def with_identifications(self):
             "identifications__user",
         )
 
+    def with_list_prefetches(self):
+        """Add prefetches the list serializer needs (detection paths, classifications).
+
+        See `ami.main.models_future.occurrence` for the prefetch definitions.
+        """
+        from ami.main.models_future.occurrence import prefetches_for_list_serializer
+
+        return self.prefetch_related(*prefetches_for_list_serializer())
+
     def with_best_detection(self):
         """
         Annotate the queryset with fields from the best detection.

ami/main/models_future/occurrence.py

@@ -0,0 +1,119 @@
+"""
+Reusable Prefetch factories for Occurrence list-view rendering.
+
+Centralising these lets the queryset, serializer, and any future caller share
+a single source of truth for what data the list view needs eagerly loaded.
+
+Tracking issue: https://github.com/RolnickLab/antenna/issues/1271
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from django.db.models import Prefetch
+
+if TYPE_CHECKING:
+    from ami.main.models import Classification, Identification, Occurrence
+
+
+def prefetch_detection_images() -> Prefetch:
+    """Detections with non-null `path`, ordered for stable image lists.
+
+    Populates `Occurrence.prefetched_detection_images` so the serializer can
+    yield image URLs without a per-occurrence query.
+    """
+    from ami.main.models import Detection
+
+    return Prefetch(
+        "detections",
+        queryset=(
+            Detection.objects.exclude(path=None)
+            .only("id", "occurrence_id", "path", "frame_num", "timestamp")
+            .order_by("frame_num", "timestamp")
+        ),
+        to_attr="prefetched_detection_images",
+    )
+
+
+def prefetch_classifications_for_best_prediction() -> Prefetch:
+    """All classifications under each occurrence's detections.
+
+    `taxon` and `algorithm` are joined so picking the best prediction in
+    Python (via BEST_MACHINE_PREDICTION_ORDER) does not lazy-load either.
+
+    The default `detections` relation manager populates `obj.detections.all()`
+    which the caller can walk to reach `.classifications.all()`.
+    """
+    from ami.main.models import Classification
+
+    return Prefetch(
+        "detections__classifications",
+        queryset=Classification.objects.select_related("taxon", "algorithm"),
+    )
+
+
+def prefetches_for_list_serializer() -> list[Prefetch]:
+    """All prefetches `OccurrenceListSerializer` needs to render without N+1.
+
+    Identifications are covered by `OccurrenceQuerySet.with_identifications()`
+    which is already applied in the list viewset; intentionally not duplicated
+    here.
+    """
+    return [
+        prefetch_detection_images(),
+        prefetch_classifications_for_best_prediction(),
+    ]
+
+
+def best_prediction_from_prefetch(occurrence: Occurrence) -> Classification | None:
+    """Pick the best machine prediction from a prefetched occurrence in Python.
+
+    Mirrors the ordering used by `BEST_MACHINE_PREDICTION_ORDER`
+    (`-terminal`, `-score`, `-pk`).
+
+    Requires `prefetch_classifications_for_best_prediction()` to have been
+    applied; walks `obj.detections.all()` -> `det.classifications.all()` from
+    the prefetch cache.
+    """
+    classifications: list = []
+    for det in occurrence.detections.all():
+        classifications.extend(det.classifications.all())
+    if not classifications:
+        return None
+    classifications.sort(
+        key=lambda c: (
+            0 if getattr(c, "terminal", False) else 1,
+            -(c.score or 0.0),
+            -c.pk,
+        )
+    )
+    return classifications[0]
+
+
+def best_identification_from_prefetch(occurrence: Occurrence) -> Identification | None:
+    """Pick the most recent non-withdrawn identification from prefetched data.
+
+    Mirrors `Occurrence.best_identification` but reads from the prefetched
+    `identifications` relation rather than issuing a fresh query.
+    """
+    best: Identification | None = None
+    for ident in occurrence.identifications.all():
+        if ident.withdrawn:
+            continue
+        if best is None or (ident.created_at and best.created_at and ident.created_at > best.created_at):
+            best = ident
+    return best
+
+
+def detection_image_urls_from_prefetch(occurrence: Occurrence, limit: int | None = None) -> list[str]:
+    """Return media URLs from the `prefetched_detection_images` to_attr list.
+
+    Requires `prefetch_detection_images()` to have been applied.
+    """
+    from ami.main.models import get_media_url
+
+    detections = getattr(occurrence, "prefetched_detection_images", [])
+    if limit is not None:
+        detections = detections[:limit]
+    return [get_media_url(det.path) for det in detections]

ami/main/tests.py

@@ -2927,6 +2927,59 @@ def test_taxa_include_occurrence_determinations_not_directly_linked(self):
         )
 
 
+class TestOccurrenceListQueryCount(APITestCase):
+    """Guard against N+1 regressions in OccurrenceViewSet.list (issue #1271).
+
+    The number of queries should be roughly constant regardless of page size:
+    a small handful of pagination/auth/permission queries plus a fixed set of
+    prefetches (occurrences, detections, classifications, identifications, etc.).
+    """
+
+    def setUp(self):
+        self.project, self.deployment = setup_test_project()
+        create_taxa(self.project)
+        create_captures(deployment=self.deployment, num_nights=1, images_per_night=10)
+
+        # Use a low threshold so all occurrences are returned
+        self.project.default_filters_score_threshold = 0.0
+        self.project.save()
+
+        self.user = User.objects.create_user(email="qcount@insectai.org", is_staff=False, is_superuser=False)
+        self.client.force_authenticate(user=self.user)
+
+    def _list_occurrences(self, limit: int) -> int:
+        from django.core.cache import caches
+        from django.test.utils import CaptureQueriesContext
+
+        url = f"/api/v2/occurrences/?project_id={self.project.pk}&limit={limit}"
+        # Clear cachalot cache between runs so query counts reflect cold state
+        try:
+            caches["default"].clear()
+        except Exception:
+            pass
+
+        with CaptureQueriesContext(connection) as ctx:
+            res = self.client.get(url)
+        self.assertEqual(res.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(res.data["results"]), min(limit, res.data["count"]))
+        return len(ctx.captured_queries)
+
+    def test_list_query_count_does_not_scale_with_page_size(self):
+        """Query count for a page of 25 should be very close to a page of 5."""
+        create_occurrences(deployment=self.deployment, num=25, determination_score=0.9)
+
+        small_count = self._list_occurrences(limit=5)
+        large_count = self._list_occurrences(limit=25)
+
+        # Allow a small constant overhead (e.g. cachalot bookkeeping) but
+        # nowhere near 5x scaling.
+        self.assertLessEqual(
+            large_count,
+            small_count + 5,
+            f"Query count grew with page size: {small_count} -> {large_count} (likely N+1 regression)",
+        )
+
+
 class TestProjectDefaultTaxaFilter(APITestCase):
     """
     Tests for project default taxa filtering (include/exclude lists).