diff --git a/ami/main/api/serializers.py b/ami/main/api/serializers.py index 6c899c2f7..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,80 @@ 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. + + 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]: + 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)}." + ) + + 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 + # 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..af3454c26 100644 --- a/ami/main/api/views.py +++ b/ami/main/api/views.py @@ -34,6 +34,7 @@ from ami.base.views import ProjectMixin from ami.main.api.schemas import limit_doc_param, project_id_doc_param from ami.main.api.serializers import TagSerializer +from ami.main.models_future.identifications import create_identifications_batch, resolve_occurrences from ami.main.models_future.occurrence import model_agreement_for_project, top_identifiers_for_project from ami.utils.requests import get_default_classification_threshold from ami.utils.storages import ConnectionTestResult @@ -62,6 +63,8 @@ update_detection_counts, ) from .serializers import ( + BulkIdentificationRequestSerializer, + BulkIdentificationResponseSerializer, ClassificationListSerializer, ClassificationSerializer, ClassificationWithTaxaSerializer, @@ -2237,6 +2240,64 @@ 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. + + Each item is saved independently: a bad item reports its error under its + own index in ``results`` while the rest of the batch still succeeds, so + a per-item problem never fails the request. Only a batch-level problem + (too many items, a duplicate occurrence, occurrences from more than one + project) rejects the whole request with a 400 serializer error instead + of the per-item ``results`` shape. See #1371. + """ + request_serializer = BulkIdentificationRequestSerializer(data=request.data) + request_serializer.is_valid(raise_exception=True) + items = request_serializer.validated_data["identifications"] + + occurrences = resolve_occurrences(items) + self._authorize_batch(request, occurrences) + results = create_identifications_batch(items, request.user, occurrences) + + created_count = sum(1 for result in results if result["status"] == "created") + return Response( + { + "created_count": created_count, + "error_count": len(items) - created_count, + "results": results, + } + ) + + 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`. + Create permission is granted per project, so one check per batch is + equivalent to one per occurrence — see `Identification.check_permission`. + """ + if not occurrences: + # Nothing resolved, so there is nothing to authorize against or to + # write; every item is reported as not found instead. + return + + 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()))) + # Check "create", not the action name: "bulk" is not in the model's CRUD + # permission map and would fall through to a permission no role grants. + if not probe.check_permission(request.user, "create"): + raise PermissionDenied("You do not have permission to identify occurrences in this project.") + class SiteViewSet(DefaultViewSet, ProjectMixin): """ diff --git a/ami/main/models.py b/ami/main/models.py index 3662b4107..406ed5e39 100644 --- a/ami/main/models.py +++ b/ami/main/models.py @@ -2912,7 +2912,13 @@ 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. The bulk + endpoint relies on that to check once per batch instead of once per + occurrence — revisit it if this path gains per-occurrence logic. See #1371. + """ import ami.users.roles as roles project = self.get_project() diff --git a/ami/main/models_future/identifications.py b/ami/main/models_future/identifications.py new file mode 100644 index 000000000..834b4cd53 --- /dev/null +++ b/ami/main/models_future/identifications.py @@ -0,0 +1,185 @@ +""" +Batch creation of identifications. + +Resolves every row a batch references in a fixed number of queries, validates +each item against the resolved batch, then saves each item inside its own +savepoint so one bad item does not abort the rest. See #1371. +""" + +import logging + +from django.contrib.auth.models import AbstractUser +from django.core import exceptions +from django.db import IntegrityError, transaction + +from ami.main.models import Classification, Identification, Occurrence, Taxon + +logger = logging.getLogger(__name__) + + +def resolve_occurrences(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 caller's permission check 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 create_identifications_batch( + items: list[dict], + user: AbstractUser, + occurrences: dict[int, Occurrence], +) -> list[dict]: + """ + Save one identification per item and report an outcome per item. + + Returns one result dict per item, in item order: ``{index, occurrence_id, + status: "created", id}`` on success, ``{index, occurrence_id, status: + "error", errors}`` on failure. A failed item never aborts the batch. + The caller is responsible for authorization; ``occurrences`` comes from + :func:`resolve_occurrences` so the permission check and the writes see the + same rows. + """ + taxa = _resolve_taxa(items) + agreed_identifications, agreed_predictions = _resolve_agreement_targets(items) + + results = [] + for index, item in enumerate(items): + errors = _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=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"]), + ) + try: + # Under ATOMIC_REQUESTS this is a savepoint, so rolling it back + # discards only this item and the rest of the batch continues. + # save() also withdraws the user's earlier identification on this + # occurrence and recomputes the determination. See #1371. + with transaction.atomic(): + identification.save() + except (IntegrityError, exceptions.ObjectDoesNotExist) as error: + # A referenced row (the occurrence, or something save() touches) + # was deleted between resolving the batch and writing this item. + 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 identification could not be saved. " + "A related record may have been deleted. Refresh and retry." + ] + }, + } + ) + continue + + results.append( + { + "index": index, + "occurrence_id": item["occurrence_id"], + "status": "created", + "id": identification.pk, + } + ) + + return results + + +def _resolve_taxa(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( + items: list[dict], +) -> tuple[dict[int, Identification], dict[int, Classification]]: + """Fetch the identifications and predictions that items claim to agree with.""" + 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( + 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 ( + # 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." + ] + + return errors diff --git a/ami/main/tests.py b/ami/main/tests.py index c9e3b0d8f..e4c56b144 100644 --- a/ami/main/tests.py +++ b/ami/main/tests.py @@ -3,12 +3,14 @@ import logging import typing from io import BytesIO +from unittest import mock from django.conf import settings from django.contrib.auth.models import AnonymousUser from django.core.files.uploadedfile import SimpleUploadedFile -from django.db import connection, models +from django.db import IntegrityError, connection, models from django.test import TestCase, override_settings +from django.test.utils import CaptureQueriesContext from django.utils import timezone from guardian.shortcuts import assign_perm, get_perms, remove_perm from PIL import Image @@ -18,6 +20,7 @@ from ami.exports.models import DataExport from ami.jobs.models import VALID_JOB_TYPES, Job +from ami.main.api.serializers import MAX_BULK_IDENTIFICATIONS from ami.main.models import ( Classification, Deployment, @@ -6667,3 +6670,569 @@ def test_scores_and_logits_counted_in_sql_including_empty(self): row = next(c for c in self.admin.get_queryset(self._request()) if c.pk == clf.pk) self.assertEqual(row.scores_count, 3) self.assertEqual(row.logits_count, 0) + + +BULK_IDENTIFICATIONS_ENDPOINT = "/api/v2/identifications/bulk/" + + +class BulkIdentificationTestCase(APITestCase): + """ + Shared fixture for the bulk identifications endpoint: one project with + several occurrences and a user per role. + + The subclasses pin the parts of the contract that are easy to break without + noticing: per-occurrence permission enforcement, the withdraw-previous and + determination-recompute side effects in ``Identification.save()``, and that + validation cost does not grow with batch size. See #1371. + """ + + 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(BULK_IDENTIFICATIONS_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_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] + 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] + + # 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/", + {"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): + """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): + 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(BULK_IDENTIFICATIONS_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( + BULK_IDENTIFICATIONS_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): + """ + The cap is what fails an oversized batch, not some other rule. + + 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): + """ + 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( + BULK_IDENTIFICATIONS_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(BULK_IDENTIFICATIONS_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 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) + queries_large = self.measure(large) + + slope = (queries_large - queries_small) / (large - small) + self.assertLessEqual( + slope, + 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.", + ) 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..004848c41 100644 --- a/ui/src/data-services/hooks/identifications/useCreateIdentifications.ts +++ b/ui/src/data-services/hooks/identifications/useCreateIdentifications.ts @@ -1,58 +1,112 @@ -import { useQueryClient } from '@tanstack/react-query' -import { API_ROUTES, SUCCESS_TIMEOUT } from 'data-services/constants' -import { useEffect, useState } from 'react' +import { useMutation, useQueryClient } from '@tanstack/react-query' +import axios from 'axios' +import { API_ROUTES, API_URL, SUCCESS_TIMEOUT } from 'data-services/constants' +import { ServerBulkIdentificationResponse } from 'data-services/models/identification' +import { getAuthHeader } from 'data-services/utils' +import { useEffect, useRef, useState } from 'react' +import { STRING, translate } from 'utils/language' +import { useUser } from 'utils/user/userContext' import { IdentificationFieldValues } from './types' -import { useCreateIdentification } from './useCreateIdentification' +import { convertToServerFieldValues } from './useCreateIdentification' +// 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 [lastAttempt, setLastAttempt] = useState() + const successResetTimeout = useRef>() - const error = numRejected - ? results.length > 1 - ? `${numRejected}/${results.length} updates were rejected, please retry.` - : 'The update was rejected, please retry.' - : undefined + 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 }) => { + // Match failures by the `index` the backend reports against each result, + // not by position in `results`, so a reordered or sparse response still + // maps each error back to the right submitted item. + const failedIndices = new Set( + data.results + .filter((result) => result.status === 'error') + .map((result) => result.index) + ) + const failed = submitted.filter((_, index) => failedIndices.has(index)) + setLastAttempt({ failed, total: submitted.length }) + queryClient.invalidateQueries([API_ROUTES.IDENTIFICATIONS]) + queryClient.invalidateQueries([API_ROUTES.OCCURRENCES]) + onSuccess?.() + if (!failed.length) { + successResetTimeout.current = setTimeout(() => reset(), SUCCESS_TIMEOUT) + } + }, + }) + // Clear the retry state when the selection changes. Keyed on the IDs, not + // their count, so swapping to a same-sized selection also clears it. + const selectionKey = occurrenceIds.join(',') useEffect(() => { - setResults(undefined) - }, [occurrenceIds.length]) + setLastAttempt(undefined) + }, [selectionKey]) + + // Cancel a pending success reset when the hook unmounts. + useEffect(() => () => clearTimeout(successResetTimeout.current), []) + + const numRejected = lastAttempt?.failed.length + const partialError = numRejected + ? lastAttempt && lastAttempt.total > 1 + ? translate(STRING.MESSAGE_IDENTIFICATIONS_REJECTED, { + numRejected, + total: lastAttempt.total, + }) + : translate(STRING.MESSAGE_IDENTIFICATION_REJECTED) + : undefined + // A rejected whole request (permission denied, invalid batch) surfaces here. + const requestError = isError + ? translate(STRING.MESSAGE_IDENTIFICATION_REJECTED) + : undefined + // The newest failure wins: a retry rejected at the request level replaces + // the partial-failure message left over from the previous attempt. + const error = requestError ?? partialError 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 + // A success reset scheduled by an earlier submission must not fire while + // this one is in flight. + clearTimeout(successResetTimeout.current) + 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. + } }, } } diff --git a/ui/src/data-services/models/identification.ts b/ui/src/data-services/models/identification.ts new file mode 100644 index 000000000..95d880631 --- /dev/null +++ b/ui/src/data-services/models/identification.ts @@ -0,0 +1,13 @@ +export interface ServerBulkIdentificationResult { + index: number + occurrence_id: number + status: 'created' | 'error' + id?: number + errors?: Record +} + +export interface ServerBulkIdentificationResponse { + created_count: number + error_count: number + results: ServerBulkIdentificationResult[] +} diff --git a/ui/src/utils/language.ts b/ui/src/utils/language.ts index 6c6b3c2be..cf5430325 100644 --- a/ui/src/utils/language.ts +++ b/ui/src/utils/language.ts @@ -191,6 +191,8 @@ export enum STRING { MESSAGE_DRAFTS, MESSAGE_EXPORT_TIP, MESSAGE_HAS_ACCOUNT, + MESSAGE_IDENTIFICATION_REJECTED, + MESSAGE_IDENTIFICATIONS_REJECTED, MESSAGE_IMAGE_FORMAT, MESSAGE_IMAGE_SIZE, MESSAGE_IMAGE_TOO_BIG, @@ -565,6 +567,10 @@ const ENGLISH_STRINGS: { [key in STRING]: string } = { [STRING.MESSAGE_EXPORT_TIP]: 'We support two export formats: one compact and easy to use, and one that includes all raw data. To include all data in the export, skip "Capture set".', [STRING.MESSAGE_HAS_ACCOUNT]: 'Already have an account?', + [STRING.MESSAGE_IDENTIFICATION_REJECTED]: + 'The update was rejected, please retry.', + [STRING.MESSAGE_IDENTIFICATIONS_REJECTED]: + '{{numRejected}}/{{total}} updates were rejected, please retry.', [STRING.MESSAGE_IMAGE_FORMAT]: 'Valid formats are PNG, GIF and JPEG.', [STRING.MESSAGE_IMAGE_SIZE]: 'The image must smaller than {{value}} {{unit}}.',