From e203c7e7e33e6363e4200f25a039609bd8a1b193 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Fri, 17 Jul 2026 04:35:56 -0700 Subject: [PATCH 1/5] feat(identifications): add bulk create endpoint Applying determinations to a page of occurrences currently costs one HTTP request per occurrence: the identification UI fans out N POSTs with Promise.allSettled and retries whichever ones were rejected. Add POST /api/v2/identifications/bulk/ so the same work is one request. The request is a list of per-item objects rather than one taxon applied to a list of occurrence IDs, because the "Agree" action sends a different taxon and a different agreement target for every occurrence. This matches the payload the frontend already builds. Each item is saved through the existing Identification.save(), so withdrawing the user's previous identification and recomputing the occurrence determination keep working exactly as they do for a single POST. Items are saved in their own savepoints and reported individually, so one stale occurrence does not discard the rest of the batch. Occurrences and taxa are resolved in one query each, so validation does not scale with the size of the batch. A batch is authorized once against the single project it belongs to; permission to create an identification depends only on the project, and the check is routed through the same Identification.check_permission call the single-item endpoint makes. Co-Authored-By: Claude --- ami/main/api/serializers.py | 66 ++++ ami/main/api/views.py | 196 +++++++++++- ami/main/models.py | 10 +- ami/main/test_bulk_identifications.py | 423 ++++++++++++++++++++++++++ 4 files changed, 693 insertions(+), 2 deletions(-) create mode 100644 ami/main/test_bulk_identifications.py diff --git a/ami/main/api/serializers.py b/ami/main/api/serializers.py index 6c899c2f7..e2dc536e4 100644 --- a/ami/main/api/serializers.py +++ b/ami/main/api/serializers.py @@ -800,6 +800,72 @@ class Meta: ] +class BulkIdentificationItemSerializer(serializers.Serializer): + """ + One identification within a bulk request. + + Related objects are declared as plain integers rather than + `PrimaryKeyRelatedField` so that the view can resolve the whole batch in a + fixed number of queries. `PrimaryKeyRelatedField` issues one query per field + per item, which would make validation alone scale with the size of the batch. + + `withdrawn` is deliberately absent: the model maintains it, and letting a + client set it would break the "one active identification per user per + occurrence" invariant. + """ + + occurrence_id = serializers.IntegerField() + taxon_id = serializers.IntegerField() + comment = serializers.CharField(required=False, allow_blank=True, default="") + agreed_with_identification_id = serializers.IntegerField(required=False, allow_null=True, default=None) + agreed_with_prediction_id = serializers.IntegerField(required=False, allow_null=True, default=None) + + +class BulkIdentificationRequestSerializer(serializers.Serializer): + """Validates the shape of a bulk request, before any occurrence is looked up.""" + + identifications = BulkIdentificationItemSerializer(many=True, allow_empty=False) + + def validate_identifications(self, value: list[dict]) -> list[dict]: + from ami.main.api.views import MAX_BULK_IDENTIFICATIONS + + if len(value) > MAX_BULK_IDENTIFICATIONS: + raise serializers.ValidationError( + f"A single request may contain at most {MAX_BULK_IDENTIFICATIONS} identifications, " + f"got {len(value)}." + ) + + occurrence_ids = [item["occurrence_id"] for item in value] + duplicates = sorted({pk for pk in occurrence_ids if occurrence_ids.count(pk) > 1}) + if duplicates: + # Two identifications for one occurrence in one batch have no defined + # winner: the outcome would depend on insert ordering rather than on + # anything the client asked for. + raise serializers.ValidationError( + f"Each occurrence may appear only once per request. Repeated occurrence IDs: {duplicates}." + ) + + return value + + +class BulkIdentificationResultSerializer(serializers.Serializer): + """The outcome of a single submitted item, matched to the request by `index`.""" + + index = serializers.IntegerField() + occurrence_id = serializers.IntegerField() + status = serializers.ChoiceField(choices=["created", "error"]) + id = serializers.IntegerField(required=False) + errors = serializers.DictField(required=False) + + +class BulkIdentificationResponseSerializer(serializers.Serializer): + """Per-item outcomes for a bulk request, in the order the items were submitted.""" + + created_count = serializers.IntegerField() + error_count = serializers.IntegerField() + results = BulkIdentificationResultSerializer(many=True) + + class TaxonDetectionsSerializer(DefaultSerializer): class Meta: model = Detection diff --git a/ami/main/api/views.py b/ami/main/api/views.py index 591a4a000..dc2a518ed 100644 --- a/ami/main/api/views.py +++ b/ami/main/api/views.py @@ -6,7 +6,7 @@ from django.contrib.postgres.search import TrigramSimilarity from django.core import exceptions from django.core.files.storage import default_storage -from django.db import models +from django.db import models, transaction from django.db.models import OuterRef, Prefetch, Q, Subquery from django.db.models.query import QuerySet from django.forms import BooleanField, CharField, IntegerField @@ -62,6 +62,8 @@ update_detection_counts, ) from .serializers import ( + BulkIdentificationRequestSerializer, + BulkIdentificationResponseSerializer, ClassificationListSerializer, ClassificationSerializer, ClassificationWithTaxaSerializer, @@ -2205,6 +2207,14 @@ def get_serializer_class(self): return PageSerializer +#: Upper bound on a single bulk identification request. +#: The write path saves each identification individually, so a batch holds a +#: database connection for roughly this many round trips. The identification UI +#: can only select occurrences on the current page, so real batches are far +#: smaller than this; the cap exists to bound the worst case, not to shape the UI. +MAX_BULK_IDENTIFICATIONS = 200 + + class IdentificationViewSet(DefaultViewSet): """ API endpoint that allows identifications to be viewed or edited. @@ -2237,6 +2247,190 @@ def perform_create(self, serializer): serializer.save(user=self.request.user) + @extend_schema( + request=BulkIdentificationRequestSerializer, + responses={200: BulkIdentificationResponseSerializer}, + ) + @action(detail=False, methods=["post"]) + def bulk(self, request): + """ + Create many identifications in one request. + + Applying determinations to a page of occurrences otherwise costs one HTTP + request per occurrence. Each item is saved independently, so a single bad + item (an occurrence deleted since the page was loaded, say) reports an + error against its own index while the rest of the batch still succeeds. + The response lists an outcome per submitted item, in request order. + """ + request_serializer = BulkIdentificationRequestSerializer(data=request.data) + request_serializer.is_valid(raise_exception=True) + items = request_serializer.validated_data["identifications"] + + occurrences = self._resolve_occurrences(items) + self._authorize_batch(request, occurrences) + taxa = self._resolve_taxa(items) + agreed_identifications, agreed_predictions = self._resolve_agreement_targets(items) + + results = [] + created_count = 0 + for index, item in enumerate(items): + errors = self._validate_item(item, occurrences, taxa, agreed_identifications, agreed_predictions) + if errors: + results.append( + { + "index": index, + "occurrence_id": item["occurrence_id"], + "status": "error", + "errors": errors, + } + ) + continue + + identification = Identification( + occurrence=occurrences[item["occurrence_id"]], + taxon=taxa[item["taxon_id"]], + user=request.user, + comment=item["comment"], + agreed_with_identification=agreed_identifications.get(item["agreed_with_identification_id"]), + agreed_with_prediction=agreed_predictions.get(item["agreed_with_prediction_id"]), + ) + # Each item commits on its own so that a failure part-way through the + # batch cannot undo the identifications already saved. save() also + # withdraws the user's earlier identification on this occurrence and + # recomputes the occurrence's determination. + with transaction.atomic(): + identification.save() + + created_count += 1 + results.append( + { + "index": index, + "occurrence_id": item["occurrence_id"], + "status": "created", + "id": identification.pk, + } + ) + + return Response( + { + "created_count": created_count, + "error_count": len(items) - created_count, + "results": results, + } + ) + + def _resolve_occurrences(self, items: list[dict]) -> dict[int, Occurrence]: + """Fetch every occurrence in the batch in one query, with its project loaded.""" + occurrence_ids = {item["occurrence_id"] for item in items} + # select_related("project") lets the permission probe and the single-project + # check read the project without a query per occurrence. + return { + occurrence.pk: occurrence + for occurrence in Occurrence.objects.filter(pk__in=occurrence_ids).select_related("project") + } + + def _authorize_batch(self, request, occurrences: dict[int, Occurrence]) -> None: + """ + Authorize the batch against the single project it belongs to. + + A `detail=False` action never passes through `has_object_permission`, so the + check has to happen here. Permission to create an identification is granted + per project, so one check per batch is equivalent to one check per + occurrence — see the note on `Identification.check_permission`. + """ + if not occurrences: + raise api_exceptions.ValidationError({"identifications": ["None of the requested occurrences exist."]}) + + projects = {occurrence.project for occurrence in occurrences.values()} + if len(projects) > 1: + raise api_exceptions.ValidationError( + {"identifications": ["All occurrences in a request must belong to the same project."]} + ) + + probe = Identification(occurrence=next(iter(occurrences.values()))) + if not probe.check_permission(request.user, "create"): + # "create", not the name of this action: the action name is not in the + # model's CRUD permission map and would silently fall through to a + # permission that no role grants. + raise PermissionDenied("You do not have permission to identify occurrences in this project.") + + def _resolve_taxa(self, items: list[dict]) -> dict[int, Taxon]: + """Fetch every taxon in the batch in one query.""" + taxon_ids = {item["taxon_id"] for item in items} + return {taxon.pk: taxon for taxon in Taxon.objects.filter(pk__in=taxon_ids)} + + def _resolve_agreement_targets( + self, items: list[dict] + ) -> tuple[dict[int, Identification], dict[int, Classification]]: + """ + Fetch the identifications and predictions that items claim to agree with. + + Both are recorded only as provenance, but they still have to be real and + belong to the occurrence being identified. + """ + identification_ids = { + item["agreed_with_identification_id"] for item in items if item["agreed_with_identification_id"] + } + prediction_ids = {item["agreed_with_prediction_id"] for item in items if item["agreed_with_prediction_id"]} + + agreed_identifications = {} + if identification_ids: + agreed_identifications = { + identification.pk: identification + for identification in Identification.objects.filter(pk__in=identification_ids) + } + + agreed_predictions = {} + if prediction_ids: + agreed_predictions = { + classification.pk: classification + for classification in Classification.objects.filter(pk__in=prediction_ids).select_related("detection") + } + + return agreed_identifications, agreed_predictions + + def _validate_item( + self, + item: dict, + occurrences: dict[int, Occurrence], + taxa: dict[int, Taxon], + agreed_identifications: dict[int, Identification], + agreed_predictions: dict[int, Classification], + ) -> dict[str, list[str]]: + """Check one item against the already-fetched batch. Returns field errors, empty when valid.""" + errors: dict[str, list[str]] = {} + + occurrence = occurrences.get(item["occurrence_id"]) + if occurrence is None: + errors["occurrence_id"] = [f"Occurrence {item['occurrence_id']} does not exist."] + + if item["taxon_id"] not in taxa: + errors["taxon_id"] = [f"Taxon {item['taxon_id']} does not exist."] + + agreed_identification_id = item["agreed_with_identification_id"] + if agreed_identification_id: + agreed = agreed_identifications.get(agreed_identification_id) + if agreed is None: + errors["agreed_with_identification_id"] = [ + f"Identification {agreed_identification_id} does not exist." + ] + elif occurrence is not None and agreed.occurrence_id != occurrence.pk: + errors["agreed_with_identification_id"] = [ + "An identification can only agree with another identification of the same occurrence." + ] + + agreed_prediction_id = item["agreed_with_prediction_id"] + if agreed_prediction_id: + prediction = agreed_predictions.get(agreed_prediction_id) + if prediction is None: + errors["agreed_with_prediction_id"] = [f"Classification {agreed_prediction_id} does not exist."] + elif occurrence is not None and prediction.detection.occurrence_id != occurrence.pk: + errors["agreed_with_prediction_id"] = [ + "An identification can only agree with a prediction of the same occurrence." + ] + + return errors + class SiteViewSet(DefaultViewSet, ProjectMixin): """ diff --git a/ami/main/models.py b/ami/main/models.py index 3662b4107..22c1197e0 100644 --- a/ami/main/models.py +++ b/ami/main/models.py @@ -2912,7 +2912,15 @@ def delete(self, *args, **kwargs): update_occurrence_determination(self.occurrence, current_determination=self.taxon) def check_permission(self, user: AbstractUser | AnonymousUser, action: str) -> bool: - """Custom permission check logic for Identification model.""" + """ + Custom permission check logic for Identification model. + + The "create" branch depends only on the occurrence's project, never on the + occurrence itself. The bulk identification endpoint relies on that: it checks + "create" once for the whole batch instead of once per occurrence. Adding + per-occurrence logic to the "create" path means that endpoint has to check + each occurrence individually. + """ import ami.users.roles as roles project = self.get_project() diff --git a/ami/main/test_bulk_identifications.py b/ami/main/test_bulk_identifications.py new file mode 100644 index 000000000..7c9554237 --- /dev/null +++ b/ami/main/test_bulk_identifications.py @@ -0,0 +1,423 @@ +""" +Tests for the bulk identifications endpoint. + +The endpoint lets an identifier apply determinations to many occurrences in one +request. The tests here pin the parts of that contract that are easy to break +without noticing: per-occurrence permission enforcement, the withdraw-previous +and determination-recompute side effects that live in ``Identification.save()``, +and the fact that validation cost must not grow with the size of the batch. +""" + +import logging + +from django.db import connection +from django.test.utils import CaptureQueriesContext +from rest_framework import status +from rest_framework.test import APITestCase + +from ami.main.models import Identification, Occurrence, Taxon +from ami.tests.fixtures.main import create_captures, create_occurrences, create_taxa, setup_test_project +from ami.users.models import User +from ami.users.roles import BasicMember, Identifier, ProjectManager + +logger = logging.getLogger(__name__) + +ENDPOINT = "/api/v2/identifications/bulk/" + + +class BulkIdentificationTestCase(APITestCase): + """Shared fixture: one project with several occurrences and a user per role.""" + + def setUp(self) -> None: + self.project, self.deployment = setup_test_project(reuse=False) + create_taxa(project=self.project) + create_captures(deployment=self.deployment) + create_occurrences(deployment=self.deployment, num=4) + + self.identifier = User.objects.create_user(email="identifier@insectai.org") # type: ignore[attr-defined] + self.basic_member = User.objects.create_user(email="basic@insectai.org") # type: ignore[attr-defined] + self.non_member = User.objects.create_user(email="stranger@insectai.org") # type: ignore[attr-defined] + self.superuser = User.objects.create_user( # type: ignore[attr-defined] + email="super@insectai.org", is_staff=True, is_superuser=True + ) + Identifier.assign_user(self.identifier, self.project) + BasicMember.assign_user(self.basic_member, self.project) + ProjectManager.assign_user(self.superuser, self.project) + + self.occurrences = list(Occurrence.objects.filter(project=self.project).exclude(determination=None)) + assert len(self.occurrences) >= 4, "Fixture must provide enough occurrences to catch per-row query growth" + + self.taxon = Taxon.objects.exclude(pk=self.occurrences[0].determination_id).first() + assert self.taxon is not None + + return super().setUp() + + def post_bulk(self, items: list[dict], user: User | None = None): + if user is not None: + self.client.force_authenticate(user=user) + return self.client.post(ENDPOINT, {"identifications": items}, format="json") + + def item(self, occurrence: Occurrence, taxon: Taxon | None = None, **extra) -> dict: + return {"occurrence_id": occurrence.pk, "taxon_id": (taxon or self.taxon).pk, **extra} + + +class TestBulkIdentificationSuccess(BulkIdentificationTestCase): + def test_creates_one_identification_per_item_and_updates_determinations(self): + """The happy path: every item becomes an Identification and moves its occurrence's determination.""" + targets = self.occurrences[:3] + response = self.post_bulk([self.item(occurrence) for occurrence in targets], user=self.identifier) + + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content) + body = response.json() + self.assertEqual(body["created_count"], 3) + self.assertEqual(body["error_count"], 0) + self.assertEqual([result["status"] for result in body["results"]], ["created"] * 3) + + for occurrence in targets: + occurrence.refresh_from_db() + self.assertEqual(occurrence.determination, self.taxon) + self.assertEqual( + Identification.objects.filter(occurrence=occurrence, user=self.identifier, withdrawn=False).count(), + 1, + ) + + def test_results_are_returned_in_request_order(self): + """Clients match results to submitted items by index, so order and index must be stable.""" + targets = self.occurrences[:3] + response = self.post_bulk([self.item(occurrence) for occurrence in targets], user=self.identifier) + + body = response.json() + self.assertEqual([result["index"] for result in body["results"]], [0, 1, 2]) + self.assertEqual( + [result["occurrence_id"] for result in body["results"]], + [occurrence.pk for occurrence in targets], + ) + + def test_comment_is_saved(self): + response = self.post_bulk( + [self.item(self.occurrences[0], comment="Wing pattern matches")], user=self.identifier + ) + + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content) + identification = Identification.objects.get(pk=response.json()["results"][0]["id"]) + self.assertEqual(identification.comment, "Wing pattern matches") + + def test_withdraws_previous_identification_by_the_same_user(self): + """ + A user has one active identification per occurrence. + + ``Identification.save()`` withdraws the user's earlier identifications on that + occurrence. A test using only fresh occurrences passes whether or not the bulk + path preserves that, so this pins it with a pre-existing identification. + """ + occurrence = self.occurrences[0] + previous = Identification.objects.create(occurrence=occurrence, taxon=self.taxon, user=self.identifier) + self.assertFalse(previous.withdrawn) + + other_taxon = Taxon.objects.exclude(pk__in=[self.taxon.pk]).first() + assert other_taxon is not None + response = self.post_bulk([self.item(occurrence, taxon=other_taxon)], user=self.identifier) + + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content) + previous.refresh_from_db() + self.assertTrue(previous.withdrawn, "The user's earlier identification must be withdrawn") + occurrence.refresh_from_db() + self.assertEqual(occurrence.determination, other_taxon) + + def test_does_not_withdraw_identifications_by_other_users(self): + """Withdrawing is scoped to the submitting user; another identifier's opinion must survive.""" + occurrence = self.occurrences[0] + other_user_id = Identification.objects.create(occurrence=occurrence, taxon=self.taxon, user=self.superuser) + + self.post_bulk([self.item(occurrence)], user=self.identifier) + + other_user_id.refresh_from_db() + self.assertFalse(other_user_id.withdrawn) + + def test_agreeing_with_the_current_determination_raises_the_score_to_the_human_score(self): + """ + Agreeing with an occurrence's existing determination keeps the taxon and lifts + the score to the human identification's score of 1.0. + + This mirrors what a single POST to /identifications/ already does, which is the + behaviour that matters: the bulk endpoint is a faster way to do the same thing, + not a different thing. `determination_score` feeds the project's score-threshold + filters, so a bulk path that left the machine score in place would quietly change + which occurrences appear in a filtered list. + """ + occurrence = self.occurrences[0] + original_taxon = occurrence.determination + self.assertEqual(occurrence.determination_score, 0.9) + + response = self.post_bulk( + [self.item(occurrence, taxon=original_taxon, agreed_with_prediction_id=occurrence.best_prediction.pk)], + user=self.identifier, + ) + + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content) + occurrence.refresh_from_db() + self.assertEqual(occurrence.determination, original_taxon) + self.assertEqual(occurrence.determination_score, 1.0) + + def test_bulk_and_single_post_leave_an_occurrence_in_the_same_state(self): + """ + The bulk endpoint must be indistinguishable from the single-item endpoint. + + Anything the bulk path reimplements or skips — withdrawing the user's previous + identification, recomputing the determination, the resulting score — shows up + here as a divergence, without this test having to name each rule. + """ + via_single, via_bulk = self.occurrences[0], self.occurrences[1] + + self.client.force_authenticate(user=self.identifier) + single_response = self.client.post( + "/api/v2/identifications/", + {"occurrence_id": via_single.pk, "taxon_id": self.taxon.pk, "comment": "same"}, + format="json", + ) + self.assertEqual(single_response.status_code, status.HTTP_201_CREATED, single_response.content) + + bulk_response = self.post_bulk([self.item(via_bulk, comment="same")], user=self.identifier) + self.assertEqual(bulk_response.status_code, status.HTTP_200_OK, bulk_response.content) + + via_single.refresh_from_db() + via_bulk.refresh_from_db() + self.assertEqual(via_bulk.determination_id, via_single.determination_id) + self.assertEqual(via_bulk.determination_score, via_single.determination_score) + + def identification_state(occurrence): + return sorted( + Identification.objects.filter(occurrence=occurrence).values_list("taxon_id", "withdrawn", "comment") + ) + + self.assertEqual(identification_state(via_bulk), identification_state(via_single)) + + def test_agreed_with_prediction_is_recorded(self): + """The agree provenance FK is stored so exports can report what the user agreed with.""" + occurrence = self.occurrences[0] + prediction = occurrence.best_prediction + response = self.post_bulk( + [self.item(occurrence, taxon=occurrence.determination, agreed_with_prediction_id=prediction.pk)], + user=self.identifier, + ) + + identification = Identification.objects.get(pk=response.json()["results"][0]["id"]) + self.assertEqual(identification.agreed_with_prediction_id, prediction.pk) + + +class TestBulkIdentificationPartialFailure(BulkIdentificationTestCase): + def test_valid_items_are_saved_when_one_item_fails(self): + """ + One bad item must not discard the rest of the batch. + + Mass identification is the point of this endpoint; failing 49 good rows because + a 50th occurrence was deleted mid-session would be worse than the N-request + version it replaces. + """ + items = [ + self.item(self.occurrences[0]), + {"occurrence_id": 9_999_999, "taxon_id": self.taxon.pk}, + self.item(self.occurrences[1]), + ] + response = self.post_bulk(items, user=self.identifier) + + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content) + body = response.json() + self.assertEqual(body["created_count"], 2) + self.assertEqual(body["error_count"], 1) + self.assertEqual([result["status"] for result in body["results"]], ["created", "error", "created"]) + self.assertIn("occurrence_id", body["results"][1]["errors"]) + + for occurrence in self.occurrences[:2]: + occurrence.refresh_from_db() + self.assertEqual(occurrence.determination, self.taxon) + + def test_unknown_taxon_is_reported_per_item(self): + items = [self.item(self.occurrences[0]), {"occurrence_id": self.occurrences[1].pk, "taxon_id": 9_999_999}] + response = self.post_bulk(items, user=self.identifier) + + body = response.json() + self.assertEqual(body["created_count"], 1) + self.assertIn("taxon_id", body["results"][1]["errors"]) + + def test_agreed_with_identification_from_another_occurrence_is_rejected(self): + """Agree provenance must point at the occurrence being identified, not an unrelated one.""" + foreign = Identification.objects.create(occurrence=self.occurrences[1], taxon=self.taxon, user=self.superuser) + response = self.post_bulk( + [self.item(self.occurrences[0], agreed_with_identification_id=foreign.pk)], + user=self.identifier, + ) + + body = response.json() + self.assertEqual(body["created_count"], 0) + self.assertIn("agreed_with_identification_id", body["results"][0]["errors"]) + + def test_a_failed_item_does_not_roll_back_successful_items(self): + """Each item commits independently; a later failure must not undo an earlier success.""" + items = [self.item(self.occurrences[0]), {"occurrence_id": self.occurrences[1].pk, "taxon_id": 9_999_999}] + self.post_bulk(items, user=self.identifier) + + self.assertTrue(Identification.objects.filter(occurrence=self.occurrences[0]).exists()) + self.assertFalse(Identification.objects.filter(occurrence=self.occurrences[1]).exists()) + + +class TestBulkIdentificationValidation(BulkIdentificationTestCase): + def test_empty_batch_is_rejected(self): + response = self.post_bulk([], user=self.identifier) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + + def test_missing_identifications_key_is_rejected(self): + self.client.force_authenticate(user=self.identifier) + response = self.client.post(ENDPOINT, {}, format="json") + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + + def test_non_integer_occurrence_id_returns_400_not_500(self): + self.client.force_authenticate(user=self.identifier) + response = self.client.post( + ENDPOINT, {"identifications": [{"occurrence_id": "abc", "taxon_id": self.taxon.pk}]}, format="json" + ) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + + def test_batch_larger_than_the_cap_is_rejected(self): + from ami.main.api.views import MAX_BULK_IDENTIFICATIONS + + items = [self.item(self.occurrences[0]) for _ in range(MAX_BULK_IDENTIFICATIONS + 1)] + response = self.post_bulk(items, user=self.identifier) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + + def test_duplicate_occurrence_ids_are_rejected(self): + """ + Two identifications for one occurrence in a single batch have no defined outcome. + + Which one wins would depend on insert-order tiebreaks rather than on anything the + client asked for, so the batch is rejected instead. + """ + response = self.post_bulk( + [self.item(self.occurrences[0]), self.item(self.occurrences[0])], user=self.identifier + ) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + + def test_batch_spanning_two_projects_is_rejected(self): + """A batch is authorized against a single project, so it must not span projects.""" + other_project, other_deployment = setup_test_project(reuse=False) + create_taxa(project=other_project) + create_captures(deployment=other_deployment) + create_occurrences(deployment=other_deployment, num=1) + Identifier.assign_user(self.identifier, other_project) + other_occurrence = Occurrence.objects.filter(project=other_project).exclude(determination=None).first() + assert other_occurrence is not None + + response = self.post_bulk([self.item(self.occurrences[0]), self.item(other_occurrence)], user=self.identifier) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + + def test_withdrawn_cannot_be_set_through_the_bulk_endpoint(self): + """`withdrawn` is managed by the model; accepting it from a client would corrupt the invariant.""" + response = self.post_bulk([self.item(self.occurrences[0], withdrawn=True)], user=self.identifier) + + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content) + identification = Identification.objects.get(pk=response.json()["results"][0]["id"]) + self.assertFalse(identification.withdrawn) + + +class TestBulkIdentificationPermissions(BulkIdentificationTestCase): + """ + The permission matrix. + + A `detail=False` action is never routed through `has_object_permission`, so the + endpoint has to run the check itself. These cases fail loudly if it stops doing so. + """ + + def test_identifier_can_create(self): + response = self.post_bulk([self.item(self.occurrences[0])], user=self.identifier) + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content) + + def test_project_manager_can_create(self): + manager = User.objects.create_user(email="manager@insectai.org") # type: ignore[attr-defined] + ProjectManager.assign_user(manager, self.project) + response = self.post_bulk([self.item(self.occurrences[0])], user=manager) + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content) + + def test_superuser_can_create(self): + response = self.post_bulk([self.item(self.occurrences[0])], user=self.superuser) + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content) + + def test_basic_member_is_forbidden(self): + """A project member without the identifier role must not be able to identify.""" + response = self.post_bulk([self.item(self.occurrences[0])], user=self.basic_member) + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + self.assertFalse(Identification.objects.filter(user=self.basic_member).exists()) + + def test_non_member_is_forbidden(self): + response = self.post_bulk([self.item(self.occurrences[0])], user=self.non_member) + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + self.assertFalse(Identification.objects.filter(user=self.non_member).exists()) + + def test_anonymous_is_rejected(self): + response = self.client.post(ENDPOINT, {"identifications": [self.item(self.occurrences[0])]}, format="json") + self.assertIn( + response.status_code, (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN), response.content + ) + self.assertFalse(Identification.objects.exists()) + + def test_permission_is_denied_for_the_whole_batch(self): + """An unauthorized batch writes nothing at all, not a partial prefix.""" + items = [self.item(occurrence) for occurrence in self.occurrences[:3]] + self.post_bulk(items, user=self.non_member) + self.assertFalse(Identification.objects.exists()) + + +class TestBulkIdentificationQueryCount(BulkIdentificationTestCase): + """ + Query cost must stay linear in the batch, with a small and stable per-item slope. + + Asserting one exact number for one batch size cannot distinguish fixed cost from + per-item cost, so it cannot catch an N+1 hidden in validation or in the response. + Measuring two batch sizes and comparing the slope can. + """ + + def measure(self, size: int) -> int: + # A fresh project per measurement keeps the two runs independent. + project, deployment = setup_test_project(reuse=False) + create_taxa(project=project) + create_captures(deployment=deployment) + create_occurrences(deployment=deployment, num=size) + Identifier.assign_user(self.identifier, project) + occurrences = list(Occurrence.objects.filter(project=project).exclude(determination=None))[:size] + taxon = Taxon.objects.exclude(pk=occurrences[0].determination_id).first() + assert taxon is not None + items = [{"occurrence_id": occurrence.pk, "taxon_id": taxon.pk} for occurrence in occurrences] + + self.client.force_authenticate(user=self.identifier) + with CaptureQueriesContext(connection) as captured: + response = self.client.post(ENDPOINT, {"identifications": items}, format="json") + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content) + self.assertEqual(response.json()["created_count"], size) + return len(captured.captured_queries) + + def test_per_item_query_cost_stays_within_budget(self): + """ + Each extra identification in a batch costs a bounded number of queries. + + Measured at the time of writing: 8 queries per item, all of them inside + `Identification.save()` (withdraw previous, insert, then + `update_occurrence_determination` reading the current determination, + `best_identification` and `best_prediction` before saving the occurrence). + Resolving occurrences and taxa is batched and costs 2 queries for the whole + request, so it does not appear in this slope. + + The budget of 10 leaves headroom for the existing write path while still + failing if the batch resolve is replaced by a per-item lookup, which would + add roughly three queries per item. + """ + small, large = 2, 6 + queries_small = self.measure(small) + queries_large = self.measure(large) + + slope = (queries_large - queries_small) / (large - small) + self.assertLessEqual( + slope, + 10, + f"Each identification should cost a bounded number of queries, measured {slope:.1f} " + f"({queries_small} queries for {small} items, {queries_large} for {large}). " + f"A jump here usually means something started querying per item.", + ) From 4927e142cf1bd073ecd7fe86ba22e8d52114ad7b Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Fri, 17 Jul 2026 04:49:31 -0700 Subject: [PATCH 2/5] fix(identifications): make bulk partial success survive a failed write The endpoint promised that one bad item would not discard the rest of the batch, but ATOMIC_REQUESTS wraps the whole request in a transaction, so the per-item atomic() block was a savepoint rather than a separate transaction and nothing committed on its own. An error raised inside save() propagated, aborted the request transaction, and rolled back every identification already made in it, returning a 500 instead of a per-item error. Catch the failure so the savepoint rolls back that item alone and the loop continues, and correct the comment that claimed otherwise. Also fixed while reviewing: - Agreeing with a prediction whose detection had been deleted raised an AttributeError and returned a 500, because Classification.detection is nullable. It is now reported as an item error. - A batch in which no occurrence existed returned a batch-level 400 while a batch in which only some were missing reported per-item errors. The same failure now answers the same way in both cases. - MAX_BULK_IDENTIFICATIONS moved to serializers, where it is enforced, which removes an import of views from inside a serializer method. Tests: the partial-success guarantee is now pinned by forcing a write to fail mid-batch, which fails without the fix above. The cap test used repeated IDs, so the duplicate check rejected the batch and the test passed with the cap removed; it now uses distinct IDs and asserts the message. The equivalence test seeded no prior identification, so it never exercised withdraw-previous. Added coverage for unknown and cross-occurrence agreement targets and for an all-missing batch, and tightened the per-item query budget to the measured cost. Co-Authored-By: Claude --- ami/main/api/serializers.py | 17 ++- ami/main/api/views.py | 59 +++++++--- ami/main/test_bulk_identifications.py | 153 ++++++++++++++++++++++++-- 3 files changed, 200 insertions(+), 29 deletions(-) diff --git a/ami/main/api/serializers.py b/ami/main/api/serializers.py index e2dc536e4..ca502da38 100644 --- a/ami/main/api/serializers.py +++ b/ami/main/api/serializers.py @@ -1,3 +1,4 @@ +import collections import datetime from django.db.models import QuerySet @@ -800,6 +801,16 @@ class Meta: ] +#: Upper bound on a single bulk identification request. +#: Every identification in a batch is written inside the request's transaction, which +#: holds row locks on each occurrence until the request finishes, so an unbounded batch +#: would block other people identifying the same occurrences for as long as it ran. The +#: identification interface can only select occurrences on the page being displayed, so +#: real batches are far smaller than this; the cap bounds the worst case rather than +#: shaping the interface. +MAX_BULK_IDENTIFICATIONS = 200 + + class BulkIdentificationItemSerializer(serializers.Serializer): """ One identification within a bulk request. @@ -827,16 +838,14 @@ class BulkIdentificationRequestSerializer(serializers.Serializer): identifications = BulkIdentificationItemSerializer(many=True, allow_empty=False) def validate_identifications(self, value: list[dict]) -> list[dict]: - from ami.main.api.views import MAX_BULK_IDENTIFICATIONS - if len(value) > MAX_BULK_IDENTIFICATIONS: raise serializers.ValidationError( f"A single request may contain at most {MAX_BULK_IDENTIFICATIONS} identifications, " f"got {len(value)}." ) - occurrence_ids = [item["occurrence_id"] for item in value] - duplicates = sorted({pk for pk in occurrence_ids if occurrence_ids.count(pk) > 1}) + counts = collections.Counter(item["occurrence_id"] for item in value) + duplicates = sorted(pk for pk, count in counts.items() if count > 1) if duplicates: # Two identifications for one occurrence in one batch have no defined # winner: the outcome would depend on insert ordering rather than on diff --git a/ami/main/api/views.py b/ami/main/api/views.py index dc2a518ed..9fecbd518 100644 --- a/ami/main/api/views.py +++ b/ami/main/api/views.py @@ -6,7 +6,7 @@ from django.contrib.postgres.search import TrigramSimilarity from django.core import exceptions from django.core.files.storage import default_storage -from django.db import models, transaction +from django.db import DatabaseError, models, transaction from django.db.models import OuterRef, Prefetch, Q, Subquery from django.db.models.query import QuerySet from django.forms import BooleanField, CharField, IntegerField @@ -2207,14 +2207,6 @@ def get_serializer_class(self): return PageSerializer -#: Upper bound on a single bulk identification request. -#: The write path saves each identification individually, so a batch holds a -#: database connection for roughly this many round trips. The identification UI -#: can only select occurrences on the current page, so real batches are far -#: smaller than this; the cap exists to bound the worst case, not to shape the UI. -MAX_BULK_IDENTIFICATIONS = 200 - - class IdentificationViewSet(DefaultViewSet): """ API endpoint that allows identifications to be viewed or edited. @@ -2294,12 +2286,36 @@ def bulk(self, request): agreed_with_identification=agreed_identifications.get(item["agreed_with_identification_id"]), agreed_with_prediction=agreed_predictions.get(item["agreed_with_prediction_id"]), ) - # Each item commits on its own so that a failure part-way through the - # batch cannot undo the identifications already saved. save() also - # withdraws the user's earlier identification on this occurrence and - # recomputes the occurrence's determination. - with transaction.atomic(): - identification.save() + try: + # ATOMIC_REQUESTS wraps the whole request in a transaction, so this + # is a savepoint rather than a separate transaction. Rolling back to + # it leaves the connection usable, which is what lets the rest of the + # batch continue after one item fails. Without catching the error + # here, a single failure would abort the request and discard every + # identification already made in it. + # + # save() also withdraws the user's earlier identification on this + # occurrence and recomputes the occurrence's determination. + with transaction.atomic(): + identification.save() + except (DatabaseError, exceptions.ObjectDoesNotExist) as error: + # The occurrence can be deleted between resolving the batch and + # writing it, which surfaces as an integrity error or a missing row + # part-way through save(). + logger.warning( + f"Bulk identification of occurrence {item['occurrence_id']} failed and was skipped: {error}" + ) + results.append( + { + "index": index, + "occurrence_id": item["occurrence_id"], + "status": "error", + "errors": { + "occurrence_id": ["This occurrence could not be identified. It may have been deleted."] + }, + } + ) + continue created_count += 1 results.append( @@ -2339,7 +2355,11 @@ def _authorize_batch(self, request, occurrences: dict[int, Occurrence]) -> None: occurrence — see the note on `Identification.check_permission`. """ if not occurrences: - raise api_exceptions.ValidationError({"identifications": ["None of the requested occurrences exist."]}) + # Nothing resolved, so there is no project to authorize against and + # nothing to write. Every item is reported as not found instead, which + # keeps a batch of one missing occurrence answering the same way as a + # batch where only some are missing. + return projects = {occurrence.project for occurrence in occurrences.values()} if len(projects) > 1: @@ -2424,7 +2444,12 @@ def _validate_item( prediction = agreed_predictions.get(agreed_prediction_id) if prediction is None: errors["agreed_with_prediction_id"] = [f"Classification {agreed_prediction_id} does not exist."] - elif occurrence is not None and prediction.detection.occurrence_id != occurrence.pk: + elif occurrence is not None and ( + # A classification keeps its row when its detection is deleted, so the + # link back to an occurrence can be missing entirely. + prediction.detection is None + or prediction.detection.occurrence_id != occurrence.pk + ): errors["agreed_with_prediction_id"] = [ "An identification can only agree with a prediction of the same occurrence." ] diff --git a/ami/main/test_bulk_identifications.py b/ami/main/test_bulk_identifications.py index 7c9554237..b89343208 100644 --- a/ami/main/test_bulk_identifications.py +++ b/ami/main/test_bulk_identifications.py @@ -9,12 +9,14 @@ """ import logging +from unittest import mock -from django.db import connection +from django.db import IntegrityError, connection from django.test.utils import CaptureQueriesContext from rest_framework import status from rest_framework.test import APITestCase +from ami.main.api.serializers import MAX_BULK_IDENTIFICATIONS from ami.main.models import Identification, Occurrence, Taxon from ami.tests.fixtures.main import create_captures, create_occurrences, create_taxa, setup_test_project from ami.users.models import User @@ -169,6 +171,13 @@ def test_bulk_and_single_post_leave_an_occurrence_in_the_same_state(self): """ via_single, via_bulk = self.occurrences[0], self.occurrences[1] + # Seed both with an earlier identification by the same user, so that + # withdraw-previous is part of what the two paths have to agree on. + other_taxon = Taxon.objects.exclude(pk=self.taxon.pk).first() + assert other_taxon is not None + for occurrence in (via_single, via_bulk): + Identification.objects.create(occurrence=occurrence, taxon=other_taxon, user=self.identifier) + self.client.force_authenticate(user=self.identifier) single_response = self.client.post( "/api/v2/identifications/", @@ -253,13 +262,117 @@ def test_agreed_with_identification_from_another_occurrence_is_rejected(self): self.assertIn("agreed_with_identification_id", body["results"][0]["errors"]) def test_a_failed_item_does_not_roll_back_successful_items(self): - """Each item commits independently; a later failure must not undo an earlier success.""" + """A rejected item must not undo the identifications already made in the batch.""" items = [self.item(self.occurrences[0]), {"occurrence_id": self.occurrences[1].pk, "taxon_id": 9_999_999}] self.post_bulk(items, user=self.identifier) self.assertTrue(Identification.objects.filter(occurrence=self.occurrences[0]).exists()) self.assertFalse(Identification.objects.filter(occurrence=self.occurrences[1]).exists()) + def test_a_database_failure_on_one_item_is_reported_without_losing_the_others(self): + """ + A write that fails inside save() costs that item only. + + The request runs in a single transaction (ATOMIC_REQUESTS), so each item is + saved inside a savepoint and its failure is caught. Without that, the first + failure would abort the transaction, discard every identification already + made in the request, and return a 500 instead of a per-item error. This is + the case that made the endpoint's partial-success promise real, so it is + pinned by forcing a failure rather than by trusting the arrangement. + """ + failing_occurrence_id = self.occurrences[1].pk + original_save = Identification.save + + def save_but_fail_for_one(identification_self, *args, **kwargs): + if identification_self.occurrence_id == failing_occurrence_id: + raise IntegrityError("simulated conflict while saving") + return original_save(identification_self, *args, **kwargs) + + items = [self.item(occurrence) for occurrence in self.occurrences[:3]] + with mock.patch.object(Identification, "save", save_but_fail_for_one): + response = self.post_bulk(items, user=self.identifier) + + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content) + body = response.json() + self.assertEqual(body["created_count"], 2) + self.assertEqual(body["error_count"], 1) + self.assertEqual([result["status"] for result in body["results"]], ["created", "error", "created"]) + + # The surviving items are really committed, not merely reported as created. + self.assertTrue(Identification.objects.filter(occurrence=self.occurrences[0]).exists()) + self.assertFalse(Identification.objects.filter(occurrence=self.occurrences[1]).exists()) + self.assertTrue(Identification.objects.filter(occurrence=self.occurrences[2]).exists()) + + def test_every_occurrence_missing_is_reported_per_item(self): + """ + A batch where nothing resolves answers like any other batch of failures. + + A batch of one deleted occurrence and a batch where only some are deleted are + the same kind of failure, so they must not return different shapes. + """ + response = self.post_bulk([{"occurrence_id": 9_999_998, "taxon_id": self.taxon.pk}], user=self.identifier) + + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content) + body = response.json() + self.assertEqual(body["created_count"], 0) + self.assertEqual(body["error_count"], 1) + self.assertIn("occurrence_id", body["results"][0]["errors"]) + + def test_unknown_agreement_targets_are_reported_per_item(self): + response = self.post_bulk( + [ + self.item(self.occurrences[0], agreed_with_identification_id=9_999_999), + self.item(self.occurrences[1], agreed_with_prediction_id=9_999_999), + ], + user=self.identifier, + ) + + body = response.json() + self.assertEqual(body["created_count"], 0) + self.assertIn("agreed_with_identification_id", body["results"][0]["errors"]) + self.assertIn("agreed_with_prediction_id", body["results"][1]["errors"]) + + def test_agreed_with_prediction_from_another_occurrence_is_rejected(self): + foreign_prediction = self.occurrences[1].best_prediction + response = self.post_bulk( + [self.item(self.occurrences[0], agreed_with_prediction_id=foreign_prediction.pk)], + user=self.identifier, + ) + + body = response.json() + self.assertEqual(body["created_count"], 0) + self.assertIn("agreed_with_prediction_id", body["results"][0]["errors"]) + + def test_a_missing_occurrence_does_not_produce_a_spurious_agreement_error(self): + """ + With no occurrence to compare against, the agreement target cannot be + cross-checked, so the item reports the missing occurrence and nothing else. + + The target here is real, so the only error that could appear alongside would + be an unwarranted "not the same occurrence" complaint. + """ + real_target = Identification.objects.create( + occurrence=self.occurrences[1], taxon=self.taxon, user=self.superuser + ) + response = self.post_bulk( + [{"occurrence_id": 9_999_997, "taxon_id": self.taxon.pk, "agreed_with_identification_id": real_target.pk}], + user=self.identifier, + ) + + errors = response.json()["results"][0]["errors"] + self.assertEqual(list(errors), ["occurrence_id"]) + + def test_an_item_reports_every_problem_it_has(self): + """Errors are reported per field, so one item with two problems names both.""" + response = self.post_bulk( + [{"occurrence_id": 9_999_997, "taxon_id": 9_999_996}], + user=self.identifier, + ) + + errors = response.json()["results"][0]["errors"] + self.assertIn("occurrence_id", errors) + self.assertIn("taxon_id", errors) + class TestBulkIdentificationValidation(BulkIdentificationTestCase): def test_empty_batch_is_rejected(self): @@ -279,11 +392,35 @@ def test_non_integer_occurrence_id_returns_400_not_500(self): self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_batch_larger_than_the_cap_is_rejected(self): - from ami.main.api.views import MAX_BULK_IDENTIFICATIONS + """ + The cap is what fails an oversized batch, not some other rule. - items = [self.item(self.occurrences[0]) for _ in range(MAX_BULK_IDENTIFICATIONS + 1)] + Repeating one occurrence 201 times would be rejected by the duplicate check + instead, and unknown IDs would be reported per item, so both would pass this + test with the cap removed. Distinct IDs plus an assertion on the message pin + the cap itself. + """ + items = [ + {"occurrence_id": 9_000_000 + offset, "taxon_id": self.taxon.pk} + for offset in range(MAX_BULK_IDENTIFICATIONS + 1) + ] response = self.post_bulk(items, user=self.identifier) + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertIn(str(MAX_BULK_IDENTIFICATIONS), str(response.json())) + + def test_a_batch_at_the_cap_is_accepted(self): + """The cap rejects what is over it, not what is exactly at it.""" + items = [ + {"occurrence_id": 9_000_000 + offset, "taxon_id": self.taxon.pk} + for offset in range(MAX_BULK_IDENTIFICATIONS) + ] + response = self.post_bulk(items, user=self.identifier) + + # Every occurrence is unknown, so each is reported as an error rather than + # the request being rejected outright. + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content) + self.assertEqual(response.json()["error_count"], MAX_BULK_IDENTIFICATIONS) def test_duplicate_occurrence_ids_are_rejected(self): """ @@ -405,9 +542,9 @@ def test_per_item_query_cost_stays_within_budget(self): Resolving occurrences and taxa is batched and costs 2 queries for the whole request, so it does not appear in this slope. - The budget of 10 leaves headroom for the existing write path while still - failing if the batch resolve is replaced by a per-item lookup, which would - add roughly three queries per item. + The budget is the measured cost, so any new per-item query fails this test. + If a change to the write path legitimately adds one, update the number here + deliberately rather than widening the budget to accommodate it. """ small, large = 2, 6 queries_small = self.measure(small) @@ -416,7 +553,7 @@ def test_per_item_query_cost_stays_within_budget(self): slope = (queries_large - queries_small) / (large - small) self.assertLessEqual( slope, - 10, + 8, f"Each identification should cost a bounded number of queries, measured {slope:.1f} " f"({queries_small} queries for {small} items, {queries_large} for {large}). " f"A jump here usually means something started querying per item.", From cfeed1d129d4232e4f02dda1d2a6b2bee1e8c045 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Fri, 17 Jul 2026 13:56:21 -0700 Subject: [PATCH 3/5] docs: bulk identifications session notes and batched-write follow-up plan [skip ci] Co-Authored-By: Claude --- ...-identifications-batched-write-followup.md | 157 ++++++++++++++++++ .../2026-07-17-bulk-identifications.md | 132 +++++++++++++++ 2 files changed, 289 insertions(+) create mode 100644 docs/claude/planning/bulk-identifications-batched-write-followup.md create mode 100644 docs/claude/sessions/2026-07-17-bulk-identifications.md diff --git a/docs/claude/planning/bulk-identifications-batched-write-followup.md b/docs/claude/planning/bulk-identifications-batched-write-followup.md new file mode 100644 index 000000000..02e0bd3a2 --- /dev/null +++ b/docs/claude/planning/bulk-identifications-batched-write-followup.md @@ -0,0 +1,157 @@ +# Batched write path for bulk identifications — follow-up design + +**Status**: design only, not built. Follow-up to PR #1371. +**Date**: 2026-07-17 + +## Verdict + +**Do not build this yet.** The v1 loop (one `Identification.save()` per item inside a caught +savepoint) is correct, reuses the only tested determination code, and costs ~8 queries per item on +batches that are bounded by the UI page size (~25–100). A batched rewrite trades that safety for a +saving that does not matter until batches grow past a page. + +**The trigger that should force it**: a "select all N occurrences matching this filter" mass-ID +action, or raising `MAX_BULK_IDENTIFICATIONS` (`ami/main/api/serializers.py`) past ~500. At n≈100 the +loop runs ~800 queries in one request; that is acceptable. At n=10 000 it is ~80 000 and holds row +locks for the whole request — not acceptable. The number below is what changes the calculus. + +Rough query counts (read from code, **not measured**): + +| n | v1 loop | batched | note | +|------|---------|---------|------| +| 25 | ~200 | ~6 | loop fine | +| 100 | ~800 | ~6 | loop fine | +| 1 000 | ~8 000 | ~7 | loop starts to hurt | +| 10 000 | ~80 000 | ~8 | loop unacceptable; batched needed | + +Batched cost is roughly constant because every step is a set operation; the loop is linear. + +## What the batched path must reproduce + +`Identification.save()` (`ami/main/models.py`) does three things per item: + +1. Withdraw the user's other non-withdrawn identifications on that occurrence + (`.filter(occurrence=, user=).exclude(pk=).update(withdrawn=True)`). +2. Insert the row. +3. `update_occurrence_determination(occurrence)` — recompute `determination` and + `determination_score`. + +The **score quirk is the spec**: `update_occurrence_determination` compares an int +(`.values("determination")`) to a Taxon instance, so it always rewrites, and agreeing lifts the score +to 1.0. A batched path must produce the identical result. Do not fix the quirk here (see the separate +ticket below); a change to the score changes which occurrences pass score-threshold filters. + +## Algorithm sketch + +Order matters: **withdraw before insert**. You cannot exclude pks that do not exist yet, so withdraw +the user's existing identifications on the affected occurrences first, then insert, then recompute. + +```python +occ_ids = [item.occurrence_id for item in items] + +with transaction.atomic(): + # 1. Withdraw the user's current identifications on these occurrences. + Identification.objects.filter(occurrence_id__in=occ_ids, user=user, withdrawn=False)\ + .update(withdrawn=True) + + # 2. Insert all new identifications at once. + created = Identification.objects.bulk_create([Identification(...) for item in items]) + + # 3. Recompute determination for every affected occurrence in a few set queries + # (see below), then bulk_update the occurrences with update_fields. +``` + +Step 3, batched. The "best identification per occurrence" is a `DISTINCT ON`: + +```sql +SELECT DISTINCT ON (occurrence_id) occurrence_id, id, taxon_id, 1.0 AS score +FROM main_identification +WHERE occurrence_id = ANY(%s) AND withdrawn = false +ORDER BY occurrence_id, created_at DESC, id DESC -- BEST_IDENTIFICATION_ORDER +``` + +Every affected occurrence now has at least one non-withdrawn identification (we just inserted one), so +`best_prediction` never has to be consulted for this batch — that removes the one genuinely per-row +query. `bulk_update(occurrences, ["determination_id", "determination_score"])` finishes it. Human +score is always 1.0 (`Identification.score`), so the score column is a constant and needs no join. + +If a future batch can also *withdraw* identifications (this one does not), the "every occurrence has a +non-withdrawn identification" assumption breaks and `best_prediction` returns — do not assume it away +without checking. + +## Partial-failure reconciliation (the crux) + +`bulk_create` is all-or-nothing: it cannot report that row 37 of 200 failed. The v1 contract is +per-row (save the good ones, report the bad). Options: + +1. **Validate everything up front, then one atomic batch.** Resolve occurrences/taxa/agree-targets and + reject any bad item *before* writing, so the only thing that can fail at write time is a concurrent + delete. Accept that a mid-flight concurrent delete turns into a whole-batch failure (rare). Cleanest + code; changes the contract from "always partial" to "partial only for pre-write validation errors". +2. **Chunked batches with per-chunk savepoints.** `bulk_create` in chunks of, say, 100; a chunk that + raises falls back to the v1 per-item loop for that chunk only. Keeps the per-row contract exactly, + at some complexity. Best if the contract must not change. +3. **Pre-flight existence check then batch, accepting the race.** Simplest, but the race window + (check → write) makes it the least honest. + +**Recommendation: option 2** if the per-row contract is load-bearing for the FE (it is today — the +existing hook reports per-item results), option 1 if the FE is being rewritten anyway and a simpler +"all valid or 400 with per-index errors" contract is acceptable. Decide *with the FE change*, not +separately. + +## Divergence risk + +Introducing a batched recompute creates a second determination code path next to +`update_occurrence_determination`. They will drift. The honest fix is to **express the scalar function +in terms of the batch function**: `update_occurrence_determination(occurrence)` becomes +`recompute_determinations([occurrence.pk])`, so there is one implementation and the single-item +endpoint exercises the same code the batch does. This should land *before or with* the batched write, +not after. + +## The test that must land first + +`update_occurrence_determination` has no tests today (`@TODO Add tests` in the source). Before swapping +loop→batched, add a differential/property test that runs both paths over the same fixture matrix and +asserts identical `(determination_id, determination_score, withdrawn flags)`: + +Fixture matrix (each row an occurrence state before the new identification): +- agree with the current prediction +- agree with an existing human identification +- brand-new taxon (no prior human ID) +- a prior identification by the **same** user (exercises withdraw-previous) +- a prior identification by a **different** user (must survive) +- an already-withdrawn identification present +- no predictions at all +- an occurrence with no detections +- a duplicate occurrence in the batch (rejected upstream, but assert the recompute is not fed it) + +The property: for every state, loop-path result == batched-path result. That test is the gate; without +it the swap is unverifiable. + +## Suggested follow-up tickets + +### Ticket A — batched write path + +> **Title**: Make bulk identifications scale to thousands of occurrences in one request +> +> Today an identifier can submit many identifications in one request, but each is written individually, +> so a very large batch runs a query per occurrence and holds database locks for the whole request. +> That is fine for the page-sized batches the interface sends now, but a future "identify everything +> matching this filter" action would send thousands at once. This replaces the per-item write with a +> set-based one (withdraw, insert, and recompute determinations in a handful of queries regardless of +> batch size), gated behind a differential test that proves the batched recompute matches the existing +> per-item behaviour exactly. Prerequisite: tests for `update_occurrence_determination`, which has none +> today. + +### Ticket B — the score-comparison quirk + +> **Title**: Decide whether agreeing with a determination should raise its score to 1.0 +> +> `update_occurrence_determination` compares a taxon ID to a taxon object, so its "nothing changed" +> branch never runs and every identification rewrites the occurrence's determination score. In practice +> this means agreeing with an existing determination lifts its score from the machine score to 1.0. +> That may be the behaviour we want — a human-confirmed occurrence arguably should score 1.0 — but it +> is currently reached by accident, and the score feeds the default score-threshold filters, so it +> quietly affects which occurrences show up in filtered lists. This ticket is to decide the intended +> behaviour and make the code express it on purpose, with a backfill plan if the decision changes +> existing scores. diff --git a/docs/claude/sessions/2026-07-17-bulk-identifications.md b/docs/claude/sessions/2026-07-17-bulk-identifications.md new file mode 100644 index 000000000..a69a787e3 --- /dev/null +++ b/docs/claude/sessions/2026-07-17-bulk-identifications.md @@ -0,0 +1,132 @@ +# Bulk identifications endpoint (PR #1371) — session notes + +**Date**: 2026-07-17 +**Branch**: `feat/bulk-identifications` (worktree off `origin/main`) +**PR**: #1371 (draft) — *Let identifiers apply identifications to many occurrences in one request* +**Author**: AI agent session + +Notes for whoever picks this up. Written for context recovery, not as a tutorial. + +## What shipped + +`POST /api/v2/identifications/bulk/` — a `detail=False` action on `IdentificationViewSet` +(`ami/main/api/views.py`, search `def bulk`). Serializers are `Bulk*` in +`ami/main/api/serializers.py`. Tests in `ami/main/test_bulk_identifications.py` (34). + +Contract: `{"identifications": [{occurrence_id, taxon_id, comment?, agreed_with_identification_id?, +agreed_with_prediction_id?}]}` → `200 {created_count, error_count, results: [{index, occurrence_id, +status, id?, errors?}]}`. + +## Findings that shaped the design (do not re-derive these) + +### The request shape was decided by the frontend, not by preference + +The mass-ID UI **already exists** and already fans out N POSTs +(`ui/src/data-services/hooks/identifications/useCreateIdentifications.ts` — `Promise.allSettled` +plus retry-only-the-rejected). Three call sites, all already building +`IdentificationFieldValues[]`: + +- `ui/src/pages/occurrences/occurrences-actions.tsx` — "Agree/Confirm". **Heterogeneous**: each + item's taxon is *that occurrence's own* determination, and each item's agree target is its own + identification or prediction. +- `ui/src/pages/occurrence-details/suggest-id/suggest-id.tsx` — one taxon, many occurrences. +- `ui/src/pages/occurrence-details/id-quick-actions/id-button.tsx` — one taxon, many occurrences. + +The Agree flow makes `{taxon_id, occurrence_ids: []}` impossible. Hence a list of per-item objects. + +### Batch size is bounded by the page, not by the project + +`ui/src/pages/occurrences/occurrences.tsx` filters `selectedItems` to occurrences on the currently +displayed page. So real batches are ~25–100. The ">100k occurrences" scaling worry does not reach +this endpoint today. It would if a "select all matching this filter" action ever ships. + +### The task brief was wrong on two points + +- **PR #1281 "Update agree logic for human identifications" is a one-line frontend fix** + (`human-identification.tsx`, +1/−1). It is not backend agree logic. +- **`user_agrees_with_identification` (`ami/main/models.py`) has zero call sites** — dead code. It + also carries an unbounded module-level `@functools.cache` keyed on model instances (stale results + + per-worker growth) if anyone ever wires it up. + +There is **no special backend agree handling**. `agreed_with_identification` / `agreed_with_prediction` +are plain nullable FKs stored as provenance; `save()` never branches on them. + +### `update_occurrence_determination` compares an int to a Taxon (pre-existing) + +`current_determination` is read via `.values("determination")`, which yields the **raw FK id**, then +compared against `top_identification.taxon`, a **model instance**. `Taxon != int` is always true, so +the guard never short-circuits, the "no update needed" branch is effectively unreachable, and every +identification rewrites `determination` and `determination_score`. + +**Verified empirically**, not inferred: a single POST agreeing with an existing determination moves +`determination_score` from 0.9 to 1.0. + +Consequence: agreeing *does* set the score to 1.0. That may be desirable, but it is reached by +accident. `determination_score` feeds project score-threshold filters +(`ami/main/models_future/filters.py`), so changing it changes which occurrences appear in filtered +lists. **Do not "fix" it as a drive-by.** The bulk endpoint deliberately reproduces today's behaviour. + +## Two bugs caught in review that are invisible in a diff + +### `ATOMIC_REQUESTS` defeats naive per-item transactions + +`config/settings/base.py:59` sets `ATOMIC_REQUESTS = True`. The whole request is therefore already +one transaction, and a per-item `transaction.atomic()` is a **savepoint**, not an independent commit. +Wrapping each item is not enough to get partial success: an exception escaping `save()` aborts the +request transaction and discards every item already written, returning a 500. + +The fix is to **catch** the failure so the savepoint rolls back that item alone and the loop +continues. Pinned by `test_a_database_failure_on_one_item_is_reported_without_losing_the_others`, +which was confirmed to fail without the fix. + +### The action name silently becomes a permission nobody has + +A `detail=False` action is never routed through `has_object_permission`, and +`ObjectPermission.has_permission` returns `True` unconditionally (`ami/base/permissions.py`), so the +action must check permission itself. If it passes `view.action` (`"bulk"`) to `check_permission`, that +misses the CRUD map in `BaseModel.check_permission` (`ami/base/models.py`) and falls through to +`check_custom_permission` → `has_perm("bulk_identification", project)` — a permission no role grants. +Superusers still pass via guardian's shortcut, so **a happy-path test written as a superuser would +never catch it**. The action maps explicitly to `"create"`. + +Related: `Identification.check_permission`'s `"create"` branch depends only on the occurrence's +project, which is why the batch is authorized once per project. A note now lives at +`Identification.check_permission` (the line a future editor touches), not only at the endpoint. + +## Verification + +- 34 tests green (`ami.main.test_bulk_identifications`), full `ami.main` suite green. +- **Live HTTP check against a real runserver** — the suite runs under `APITestCase`, which wraps each + test in a transaction, so `ATOMIC_REQUESTS` degrades to a savepoint there and real commit behaviour + is never exercised. The live check confirmed a mixed batch returns 200 with 2 created + 1 error, the + survivors **commit** (re-read from a separate process), the prior identification is withdrawn, and + 403/400 paths behave. Scripts were kept out of the repo (throwaway). + +### Gotcha: the CI stack's *dev* database is polluted + +`docker compose -f docker-compose.ci.yml` → the `ami` database has a `main_project.license` NOT NULL +column that **exists in no migration on `main`**, left behind by another branch. `migrate` reports +"no migrations to apply" because the other branch's migration row is recorded. Seeding real data there +fails with `null value in column "license"`. + +Workaround used: create a throwaway database, migrate it, run, drop it. Build the URL from +`POSTGRES_*` env vars **inside** the container so no credential ever appears in a command. + +### Gotcha: two concurrent `--keepdb` runs corrupt each other + +Running two test commands against the CI stack at once produces spurious failures — they share the +`test_ami` database. Also hit the known cachalot issue: `--keepdb` plus permission migrations → +`IntegrityError on auth_group_permissions`. Fix is to drop the test DB **and** `redis-cli FLUSHALL` +together (see `reference_ci_compose_cachalot_flush`). + +## Follow-ups + +- **Frontend switch** — not in this PR. `useCreateIdentifications.ts` should POST the array it already + builds instead of fanning out; its retry-by-index logic must be rewritten either way, since + `PromiseSettledResult[]` has no meaning with one request. +- **Batched write path** — planned; the request contract does not change. See the planning doc. +- **`update_occurrence_determination` int-vs-Taxon comparison** — needs a decision, then tests, before + anyone touches determination logic. +- **`user_agrees_with_identification`** — delete or fix the `functools.cache`. +- **Single-item endpoint asymmetries** — it accepts a writable `withdrawn` and unvalidated + `agreed_with_*` targets. The bulk endpoint rejects both. From e4c7f70258fdd79ebc2370d3e7f202d5013fb332 Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Fri, 17 Jul 2026 14:03:50 -0700 Subject: [PATCH 4/5] feat(ui): send mass identifications in one request The occurrences list already lets an identifier select a page of occurrences and apply a determination to all of them at once ("select all", then Agree or Suggest ID). Until now that fanned out one HTTP request per occurrence, so confirming a page of twenty sent twenty requests, each doing its own authorization and its own determination recompute. During a mass-identification session that multiplies the load on the site. This points the bulk identification hook at the new POST /api/v2/identifications/bulk/ endpoint, so the whole selection goes in a single request. The three callers (Agree, Suggest ID, and the quick ID buttons) are unchanged: the hook keeps the same interface, including reporting how many items were rejected and resending only those on a retry. A partial failure is still a successful request, so the "Confirmed" state only shows once every item landed. The backend accepts occurrence and taxon IDs as strings, which is what the form sends; a test pins that contract. Co-Authored-By: Claude --- ami/main/test_bulk_identifications.py | 18 +++ .../useCreateIdentification.ts | 2 +- .../useCreateIdentifications.ts | 124 ++++++++++++------ 3 files changed, 104 insertions(+), 40 deletions(-) diff --git a/ami/main/test_bulk_identifications.py b/ami/main/test_bulk_identifications.py index b89343208..f8ecb0b36 100644 --- a/ami/main/test_bulk_identifications.py +++ b/ami/main/test_bulk_identifications.py @@ -83,6 +83,24 @@ def test_creates_one_identification_per_item_and_updates_determinations(self): 1, ) + def test_accepts_ids_sent_as_strings(self): + """ + The frontend sends occurrence and taxon IDs as JSON strings. + + The request body is built from the identification form, where IDs are + strings, so the endpoint has to accept "123" and not only 123. This pins + the frontend-to-backend contract; a stricter integer-only field would 400 + every real bulk identification. + """ + occurrence = self.occurrences[0] + response = self.post_bulk( + [{"occurrence_id": str(occurrence.pk), "taxon_id": str(self.taxon.pk)}], + user=self.identifier, + ) + + self.assertEqual(response.status_code, status.HTTP_200_OK, response.content) + self.assertEqual(response.json()["created_count"], 1) + def test_results_are_returned_in_request_order(self): """Clients match results to submitted items by index, so order and index must be stable.""" targets = self.occurrences[:3] diff --git a/ui/src/data-services/hooks/identifications/useCreateIdentification.ts b/ui/src/data-services/hooks/identifications/useCreateIdentification.ts index a35fc1ad1..ace5c984c 100644 --- a/ui/src/data-services/hooks/identifications/useCreateIdentification.ts +++ b/ui/src/data-services/hooks/identifications/useCreateIdentification.ts @@ -5,7 +5,7 @@ import { getAuthHeader } from 'data-services/utils' import { useUser } from 'utils/user/userContext' import { IdentificationFieldValues } from './types' -const convertToServerFieldValues = ( +export const convertToServerFieldValues = ( fieldValues: IdentificationFieldValues ) => ({ agreed_with_identification_id: fieldValues.agreeWith?.identificationId, diff --git a/ui/src/data-services/hooks/identifications/useCreateIdentifications.ts b/ui/src/data-services/hooks/identifications/useCreateIdentifications.ts index f7e64e04f..8a4073787 100644 --- a/ui/src/data-services/hooks/identifications/useCreateIdentifications.ts +++ b/ui/src/data-services/hooks/identifications/useCreateIdentifications.ts @@ -1,58 +1,104 @@ -import { useQueryClient } from '@tanstack/react-query' -import { API_ROUTES, SUCCESS_TIMEOUT } from 'data-services/constants' +import { useMutation, useQueryClient } from '@tanstack/react-query' +import axios from 'axios' +import { API_ROUTES, API_URL, SUCCESS_TIMEOUT } from 'data-services/constants' +import { getAuthHeader } from 'data-services/utils' import { useEffect, useState } from 'react' +import { useUser } from 'utils/user/userContext' import { IdentificationFieldValues } from './types' -import { useCreateIdentification } from './useCreateIdentification' +import { convertToServerFieldValues } from './useCreateIdentification' +interface BulkIdentificationResult { + index: number + occurrence_id: number + status: 'created' | 'error' + id?: number + errors?: Record +} + +interface BulkIdentificationResponse { + created_count: number + error_count: number + results: BulkIdentificationResult[] +} + +// Records the outcome of the last submission so a retry can resend only the +// items that failed, and so the error message can report how many did. +interface LastAttempt { + failed: IdentificationFieldValues[] + total: number +} + +/** + * Apply an identification to many occurrences in a single request. + * + * The identifications are sent together to the bulk endpoint, which reports a + * result per item, rather than one request per occurrence. A partial failure + * (an occurrence deleted since the page loaded, say) leaves the successful items + * saved and lets a retry resend only the ones that failed. + */ export const useCreateIdentifications = ( occurrenceIds: string[], onSuccess?: () => void ) => { + const { user } = useUser() const queryClient = useQueryClient() - const [results, setResults] = useState[]>() - const { createIdentification, isLoading, isSuccess, reset } = - useCreateIdentification(() => { - setTimeout(() => { - reset() - }, SUCCESS_TIMEOUT) - }, false) - - const numRejected = results?.filter( - (result) => result.status === 'rejected' - ).length - - const error = numRejected - ? results.length > 1 - ? `${numRejected}/${results.length} updates were rejected, please retry.` - : 'The update was rejected, please retry.' - : undefined + const [lastAttempt, setLastAttempt] = useState() + const { mutateAsync, isLoading, isSuccess, isError, reset } = useMutation({ + mutationFn: async (values: IdentificationFieldValues[]) => { + const { data } = await axios.post( + `${API_URL}/${API_ROUTES.IDENTIFICATIONS}/bulk/`, + { identifications: values.map(convertToServerFieldValues) }, + { headers: getAuthHeader(user) } + ) + return { data, submitted: values } + }, + onSuccess: ({ data, submitted }) => { + const failed = submitted.filter( + (_, index) => data.results[index]?.status === 'error' + ) + setLastAttempt({ failed, total: submitted.length }) + queryClient.invalidateQueries([API_ROUTES.IDENTIFICATIONS]) + queryClient.invalidateQueries([API_ROUTES.OCCURRENCES]) + onSuccess?.() + if (!failed.length) { + setTimeout(() => reset(), SUCCESS_TIMEOUT) + } + }, + }) + + // Clear the retry state when the selection changes. useEffect(() => { - setResults(undefined) + setLastAttempt(undefined) }, [occurrenceIds.length]) + const numRejected = lastAttempt?.failed.length + const partialError = numRejected + ? lastAttempt && lastAttempt.total > 1 + ? `${numRejected}/${lastAttempt.total} updates were rejected, please retry.` + : 'The update was rejected, please retry.' + : undefined + // A rejected whole request (permission denied, invalid batch) surfaces here. + const requestError = isError + ? 'The update was rejected, please retry.' + : undefined + const error = partialError ?? requestError + return { + // A partial failure is still a successful request, so only report success + // once every item landed. + isSuccess: isSuccess && !error, isLoading, - isSuccess, error, createIdentifications: async (params: IdentificationFieldValues[]) => { - const promises = params - .filter((_, index) => { - if (error) { - // Only retry rejected requests - return results?.[index]?.status === 'rejected' - } - - return true - }) - .map((variables) => createIdentification(variables)) - - setResults(undefined) - const result = await Promise.allSettled(promises) - setResults(result) - queryClient.invalidateQueries([API_ROUTES.IDENTIFICATIONS]) - queryClient.invalidateQueries([API_ROUTES.OCCURRENCES]) - onSuccess?.() + // On a retry, resend only the items that failed last time. + const toSubmit = partialError && lastAttempt ? lastAttempt.failed : params + try { + await mutateAsync(toSubmit) + } catch { + // A rejected request is surfaced through `error`; swallow it here so the + // caller's click handler does not see an unhandled rejection. + } }, } } From 7df0b5bea60f5a1f23fb24312db87b4cc7c9f42b Mon Sep 17 00:00:00 2001 From: Michael Bunsen Date: Fri, 17 Jul 2026 14:06:28 -0700 Subject: [PATCH 5/5] docs: note frontend switch landed in the bulk-identifications PR [skip ci] Co-Authored-By: Claude --- .../sessions/2026-07-17-bulk-identifications.md | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/docs/claude/sessions/2026-07-17-bulk-identifications.md b/docs/claude/sessions/2026-07-17-bulk-identifications.md index a69a787e3..9dbd0224e 100644 --- a/docs/claude/sessions/2026-07-17-bulk-identifications.md +++ b/docs/claude/sessions/2026-07-17-bulk-identifications.md @@ -119,11 +119,22 @@ Running two test commands against the CI stack at once produces spurious failure `IntegrityError on auth_group_permissions`. Fix is to drop the test DB **and** `redis-cli FLUSHALL` together (see `reference_ci_compose_cachalot_flush`). +## Frontend switch (in this PR) + +The pressure relief only lands when the FE stops fanning out N POSTs. `useCreateIdentifications.ts` now +makes one POST to `/bulk/`; the singular `useCreateIdentification.ts` (still used by the per-occurrence +Agree card in `agree.tsx`) is untouched. The plural hook keeps its return shape +(`isLoading`/`isSuccess`/`error`/`createIdentifications`), so the three callers (`occurrences-actions.tsx`, +`suggest-id.tsx`, `id-button.tsx`) are unchanged. Two behaviours preserved on purpose: retry resends +only the rejected items, and a partial failure is not reported as success (drives the Agree +"Confirmed" state). "Select all" is page-bounded (`occurrence-gallery.tsx:173` maps the current page's +`items`; default page size 20), so batches stay well under `MAX_BULK_IDENTIFICATIONS = 200`. + +FE verified: `tsc --noEmit`, eslint, prettier, and `yarn build` all clean. BE string-ID contract +(`"123"` coerced by DRF IntegerField) pinned by `test_accepts_ids_sent_as_strings`. + ## Follow-ups -- **Frontend switch** — not in this PR. `useCreateIdentifications.ts` should POST the array it already - builds instead of fanning out; its retry-by-index logic must be rewritten either way, since - `PromiseSettledResult[]` has no meaning with one request. - **Batched write path** — planned; the request contract does not change. See the planning doc. - **`update_occurrence_determination` int-vs-Taxon comparison** — needs a decision, then tests, before anyone touches determination logic.