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,63 @@ 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 `detections` cache populated by
+        `prefetch_detections_for_list()`; falls back to the model method for
+        callers that did not apply the prefetch (e.g. signals/exports).
+        """
+        from ami.main.models_future.occurrence import detection_image_urls_from_prefetch
+
+        if "detections" in getattr(obj, "_prefetched_objects_cache", {}):
+            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.
+
+        The list path prefetches detections AND nested classifications via
+        `prefetch_detections_for_list()`. The detail path only prefetches
+        detections, so calling `best_prediction_from_prefetch()` there would
+        walk `det.classifications.all()` lazily and reintroduce an N+1.
+        Gate strictly on classifications also being prefetched.
+        """
+        from ami.main.models_future.occurrence import best_prediction_from_prefetch, has_prefetched_classifications
 
+        if has_prefetched_classifications(obj):
+            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 +1439,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

@@ -60,6 +60,11 @@
 # over non-terminal, then highest score, with pk as the deterministic tiebreaker.
 BEST_MACHINE_PREDICTION_ORDER: Final = ("-terminal", "-score", "-pk")
 
+# Ordering for "best identification" selection used by Occurrence.best_identification,
+# OccurrenceQuerySet.with_verification_info(), and best_identification_from_prefetch().
+# Most recent non-withdrawn identification wins, with pk as the deterministic tiebreaker.
+BEST_IDENTIFICATION_ORDER: Final = ("-created_at", "-pk")
+
 
 class TaxonRank(OrderedEnum):
     KINGDOM = "KINGDOM"
@@ -2932,6 +2937,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.
@@ -3014,7 +3028,7 @@ def with_verification_info(self):
         """
         best_identification_subquery = Identification.objects.filter(
             occurrence=OuterRef("pk"), withdrawn=False
-        ).order_by("-created_at")
+        ).order_by(*BEST_IDENTIFICATION_ORDER)
 
         return self.annotate(
             verified_by_name=models.Subquery(best_identification_subquery.values("user__name")[:1]),
@@ -3240,7 +3254,11 @@ def best_identification(self):
 
         @TODO this could use a confidence level chosen manually by the users/experts.
         """
-        return Identification.objects.filter(occurrence=self, withdrawn=False).order_by("-created_at").first()
+        return (
+            Identification.objects.filter(occurrence=self, withdrawn=False)
+            .order_by(*BEST_IDENTIFICATION_ORDER)
+            .first()
+        )
 
     def get_determination_score(self) -> float | None:
         if not self.determination:

ami/main/models_future/occurrence.py

@@ -0,0 +1,141 @@
+"""
+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_detections_for_list() -> Prefetch:
+    """Single detections prefetch covering image URL listing AND best-prediction selection.
+
+    One pass loads detections (ordered for stable image lists) plus their
+    classifications with `taxon`/`algorithm` joined. The serializer derives
+    image URLs by filtering `path is not None` in Python and picks the best
+    machine prediction from the same cache.
+
+    Replaces the previous pair (a `to_attr` filtered list for image paths
+    plus a separate `detections__classifications` prefetch) which loaded the
+    detections relation twice.
+    """
+    from ami.main.models import Classification, Detection
+
+    return Prefetch(
+        "detections",
+        queryset=(
+            Detection.objects.prefetch_related(
+                Prefetch(
+                    "classifications",
+                    queryset=Classification.objects.select_related("taxon", "algorithm"),
+                )
+            ).order_by("frame_num", "timestamp")
+        ),
+    )
+
+
+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_detections_for_list()]
+
+
+def has_prefetched_classifications(occurrence: Occurrence) -> bool:
+    """Return True iff `detections` AND each detection's `classifications` are prefetched.
+
+    The list path prefetches both via `prefetch_detections_for_list()`. The
+    detail path prefetches only `detections` — calling
+    `best_prediction_from_prefetch()` there would walk `det.classifications.all()`
+    and reintroduce an N+1 (one query per detection).
+    """
+    cache = getattr(occurrence, "_prefetched_objects_cache", {})
+    detections = cache.get("detections")
+    if detections is None:
+        return False
+    return all("classifications" in getattr(det, "_prefetched_objects_cache", {}) for det in detections)
+
+
+def best_prediction_from_prefetch(occurrence: Occurrence) -> Classification | None:
+    """Pick the best machine prediction from a prefetched occurrence in Python.
+
+    Mirrors `Occurrence.best_prediction`, which calls `Occurrence.predictions()`
+    (per-algorithm max-score filtering) and then orders by `-terminal, -score`.
+    Replicating that grouping in Python keeps list and detail responses
+    consistent for occurrences whose top-scoring classification is non-terminal.
+
+    Requires `prefetch_detections_for_list()` (or equivalent) to have been
+    applied; walks `obj.detections.all()` -> `det.classifications.all()` from
+    the prefetch cache.
+    """
+    classifications = [c for det in occurrence.detections.all() for c in det.classifications.all()]
+    if not classifications:
+        return None
+
+    max_score_per_algo: dict[object, float] = {}
+    for c in classifications:
+        score = c.score if c.score is not None else float("-inf")
+        existing = max_score_per_algo.get(c.algorithm_id)
+        if existing is None or score > existing:
+            max_score_per_algo[c.algorithm_id] = score
+
+    candidates = [
+        c
+        for c in classifications
+        if (c.score if c.score is not None else float("-inf")) == max_score_per_algo[c.algorithm_id]
+    ]
+    if not candidates:
+        return None
+    candidates.sort(
+        key=lambda c: (
+            0 if getattr(c, "terminal", False) else 1,
+            -(c.score or 0.0),
+            -c.pk,
+        )
+    )
+    return candidates[0]
+
+
+def best_identification_from_prefetch(occurrence: Occurrence) -> Identification | None:
+    """Pick the most recent non-withdrawn identification from prefetched data.
+
+    Mirrors `Occurrence.best_identification`, which uses
+    `BEST_IDENTIFICATION_ORDER = ("-created_at", "-pk")`. `created_at=None` is
+    treated as lower than any real timestamp.
+    """
+    best: Identification | None = None
+    best_key: tuple[bool, object, int] | None = None
+    for ident in occurrence.identifications.all():
+        if ident.withdrawn:
+            continue
+        ident_key: tuple[bool, object, int] = (ident.created_at is not None, ident.created_at, ident.pk)
+        if best_key is None or ident_key > best_key:
+            best = ident
+            best_key = ident_key
+    return best
+
+
+def detection_image_urls_from_prefetch(occurrence: Occurrence, limit: int | None = None) -> list[str]:
+    """Return media URLs for the prefetched detections (filtering out `path=None`).
+
+    Requires `prefetch_detections_for_list()` to have been applied.
+    """
+    from ami.main.models import get_media_url
+
+    detections = [det for det in occurrence.detections.all() if det.path]
+    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,57 @@ 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.
+        # Failures must propagate — a warm cache hides query-scaling regressions.
+        caches["default"].clear()
+
+        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).