Fix: Update attachments in BE instead of cloning record

specifyweb/backend/stored_queries/batch_edit.py

@@ -65,7 +65,6 @@ def get_readonly_fields(table: Table):
         rel.name
         for rel in table.relationships
         if rel.relatedModelName.lower() in BATCH_EDIT_READONLY_TABLES
-        or "attachment" in rel.relatedModelName.lower()
     ]
     if table.name.lower() == "determination":
         relationships = ["preferredtaxon"]
@@ -837,6 +836,7 @@ def _relationship_is_editable(name, value):
                 overrideScope=None,
                 wbcols=wb_cols,
                 static={},
+                preserveIdentity=(base_table.name.lower() == "attachment"),
                 # FEAT: Remove this restriction to allow adding brand new data anywhere
                 # that's about the best we can do, to make relationships readonly. we can't really omit them during headers finding, because they are "still" there
                 toOne=Func.remove_keys(to_one_upload_tables, _relationship_is_editable),

specifyweb/backend/workbench/upload/scoping.py

@@ -319,6 +319,7 @@ def _backref(key):
         static=ut.static,
         toOne=to_ones,
         toMany=to_many,  # type: ignore
+        preserveIdentity=ut.preserveIdentity,
         scopingAttrs=scoping_relationships(collection, table),
         disambiguation=None,
         # Often, we'll need to recur down to clone (nested one-to-ones). Having this entire is handy in such a case

specifyweb/backend/workbench/upload/tests/test_batch_edit_table.py

@@ -1,7 +1,8 @@
+import json
 from typing import Literal
 from specifyweb.specify.utils.func import Func
 from specifyweb.specify.tests.test_api import get_table
-from specifyweb.backend.stored_queries.batch_edit import run_batch_edit_query  # type: ignore
+from specifyweb.backend.stored_queries.batch_edit import make_dataset, run_batch_edit_query  # type: ignore
 from specifyweb.backend.stored_queries.queryfield import QueryField, fields_from_json
 from specifyweb.backend.stored_queries.queryfieldspec import QueryFieldSpec
 from specifyweb.backend.stored_queries.tests.test_batch_edit import props_builder
@@ -12,7 +13,7 @@
     BatchEditPrefs,
 )
 from specifyweb.backend.workbench.upload.tests.base import UploadTestsBase
-from specifyweb.backend.workbench.upload.upload import do_upload
+from specifyweb.backend.workbench.upload.upload import do_upload, do_upload_dataset
 from specifyweb.backend.workbench.upload import auditcodes
 from specifyweb.backend.workbench.upload.upload_result import (
     Deleted,
@@ -28,6 +29,7 @@
 )
 from specifyweb.backend.workbench.upload.upload_table import UploadTable
 from specifyweb.backend.workbench.views import regularize_rows
+from specifyweb.backend.workbench.models import Spdataset
 from ..upload_plan_schema import parse_column_options, parse_plan, schema
 
 from jsonschema import validate  # type: ignore
@@ -36,13 +38,15 @@
     Spauditlogfield,
     Collectionobject,
     Agent,
+    Attachment,
     Preptype,
     Preparation,
     Collectingeventattribute,
     Collectingevent,
     Address,
     Agentspecialty,
     Collector,
+    Collectionobjectattachment,
 )
 
 lookup_in_auditlog = lambda model, _id: get_table("Spauditlog").objects.filter(
@@ -506,6 +510,41 @@ def setUp(self):
             preptype=self.preptype,
         )
 
+        self.co_1_attachment = Attachment.objects.create(
+            origfilename="co_1.jpg",
+            attachmentlocation="co_1.jpg",
+            title="CO 1 Attachment",
+        )
+        self.co_2_attachment = Attachment.objects.create(
+            origfilename="co_2.jpg",
+            attachmentlocation="co_2.jpg",
+            title="CO 2 Attachment",
+        )
+        self.co_3_attachment = Attachment.objects.create(
+            origfilename="co_3.jpg",
+            attachmentlocation="co_3.jpg",
+            title="CO 3 Attachment",
+        )
+
+        self.co_1_attachment_link = Collectionobjectattachment.objects.create(
+            collectionmemberid=self.collection.id,
+            collectionobject=self.co_1,
+            attachment=self.co_1_attachment,
+            ordinal=0,
+        )
+        self.co_2_attachment_link = Collectionobjectattachment.objects.create(
+            collectionmemberid=self.collection.id,
+            collectionobject=self.co_2,
+            attachment=self.co_2_attachment,
+            ordinal=0,
+        )
+        self.co_3_attachment_link = Collectionobjectattachment.objects.create(
+            collectionmemberid=self.collection.id,
+            collectionobject=self.co_3,
+            attachment=self.co_3_attachment,
+            ordinal=0,
+        )
+
     def _build_props(self, query_fields, base_table):
         raw = self.build_props(query_fields, base_table)
         raw["session_maker"] = self.__class__.test_session_context
@@ -570,6 +609,158 @@ def test_no_op(self):
         # We didn't change anything, nothing should change. verify just that
         list([self.enforcer(result) for result in results])
 
+    def test_batch_edit_attachment_updates_existing_record(self):
+        query_paths = [
+            ["catalognumber"],
+            ["collectionobjectattachments", "attachment", "title"],
+        ]
+
+        (headers, rows, pack, plan) = self.query_to_results(
+            *self.make_query_fields("collectionobject", query_paths)
+        )
+
+        self.assertEqual(
+            pack[0]["to_many"]["collectionobjectattachments"][0]["to_one"]["attachment"]["self"]["id"],
+            self.co_1_attachment.id,
+        )
+        self.assertIn(
+            "attachment",
+            plan.toMany["collectionobjectattachments"][0].toOne,
+        )
+        self.assertTrue(plan.toMany["collectionobjectattachments"][0].toOne["attachment"].preserveIdentity)
+
+        data = [dict(zip(headers, row)) for row in rows]
+        data[0]["Attachment title"] = "Updated CO 1 Attachment"
+
+        initial_attachment_count = Attachment.objects.count()
+
+        results = do_upload(
+            self.collection,
+            data,
+            plan,
+            self.agent.id,
+            batch_edit_packs=pack,
+        )
+
+        self.assertIsInstance(results[0].record_result, NoChange)
+
+        co_attachment_result = results[0].toMany["collectionobjectattachments"][0]
+        self.assertIsInstance(co_attachment_result.record_result, NoChange)
+
+        attachment_result = co_attachment_result.toOne["attachment"].record_result
+        self.assertIsInstance(attachment_result, Updated)
+        self.assertEqual(attachment_result.get_id(), self.co_1_attachment.id)
+
+        self.co_1_attachment.refresh_from_db()
+        self.co_1_attachment_link.refresh_from_db()
+
+        self.assertEqual(self.co_1_attachment.title, "Updated CO 1 Attachment")
+        self.assertEqual(
+            self.co_1_attachment_link.attachment_id, self.co_1_attachment.id
+        )
+        self.assertEqual(Attachment.objects.count(), initial_attachment_count)
+        self.enforce_in_log(self.co_1_attachment.id, "attachment", "UPDATE")
+
+    def test_batch_edit_attachment_table_updates_existing_record(self):
+        query_paths = [
+            ["title"],
+            ["origfilename"],
+        ]
+
+        (headers, rows, pack, plan) = self.query_to_results(
+            *self.make_query_fields("attachment", query_paths)
+        )
+
+        attachment_ids = [row_pack["self"]["id"] for row_pack in pack]
+        self.assertIn(self.co_1_attachment.id, attachment_ids)
+        self.assertTrue(plan.preserveIdentity)
+
+        row_index = attachment_ids.index(self.co_1_attachment.id)
+        data = [dict(zip(headers, row)) for row in rows]
+        data[row_index]["Attachment title"] = "Updated via Attachment Query"
+
+        initial_attachment_count = Attachment.objects.count()
+
+        results = do_upload(
+            self.collection,
+            data,
+            plan,
+            self.agent.id,
+            batch_edit_packs=pack,
+        )
+
+        attachment_result = results[row_index].record_result
+        self.assertIsInstance(attachment_result, Updated)
+        self.assertEqual(attachment_result.get_id(), self.co_1_attachment.id)
+
+        self.co_1_attachment.refresh_from_db()
+
+        self.assertEqual(self.co_1_attachment.title, "Updated via Attachment Query")
+        self.assertEqual(Attachment.objects.count(), initial_attachment_count)
+        self.enforce_in_log(self.co_1_attachment.id, "attachment", "UPDATE")
+
+    def test_batch_edit_attachment_dataset_commit_updates_original_record(self):
+        query_paths = [
+            ["catalognumber"],
+            ["collectionobjectattachments", "attachment", "title"],
+        ]
+
+        query_fields = [
+            self.make_query(QueryFieldSpec.from_path(path), 0)
+            for path in [("Collectionobject", *path) for path in query_paths]
+        ]
+        props = self._build_props(query_fields, "Collectionobject")
+
+        (headers, rows, packs, plan_json, visual_order) = run_batch_edit_query(props)
+
+        mapped_rows = [
+            [*row, json.dumps({"batch_edit": pack})] for (row, pack) in zip(rows, packs)
+        ]
+        regularized_rows = regularize_rows(len(headers), mapped_rows, skip_empty=False)
+        row_index = next(
+            index
+            for index, pack in enumerate(packs)
+            if pack["self"]["id"] == self.co_1.id
+        )
+
+        dataset_rows = [row[:] for row in regularized_rows]
+        attachment_title_index = headers.index("Attachment title")
+        dataset_rows[row_index][attachment_title_index] = "Updated CO 1 Attachment"
+
+        dataset_id, _ = make_dataset(
+            user=self.specifyuser,
+            collection=self.collection,
+            name="attachment-batch-edit",
+            headers=headers,
+            regularized_rows=dataset_rows,
+            agent=self.agent,
+            json_upload_plan=plan_json,
+            visual_order=visual_order,
+        )
+
+        dataset = Spdataset.objects.get(id=dataset_id)
+
+        initial_attachment_count = Attachment.objects.count()
+        results = do_upload_dataset(
+            self.collection,
+            self.agent.id,
+            dataset,
+            no_commit=False,
+            allow_partial=False,
+        )
+
+        attachment_result = results[row_index].toMany["collectionobjectattachments"][0].toOne["attachment"].record_result
+        self.assertIsInstance(attachment_result, Updated)
+        self.assertEqual(attachment_result.get_id(), self.co_1_attachment.id)
+
+        self.co_1_attachment.refresh_from_db()
+        self.co_1_attachment_link.refresh_from_db()
+
+        self.assertEqual(self.co_1_attachment.title, "Updated CO 1 Attachment")
+        self.assertEqual(self.co_1_attachment_link.attachment_id, self.co_1_attachment.id)
+        self.assertEqual(Attachment.objects.count(), initial_attachment_count)
+        self.enforce_in_log(self.co_1_attachment.id, "attachment", "UPDATE")
+
     def enforce_in_log(
         self,
         record_id,

specifyweb/backend/workbench/upload/upload_plan_schema.py

@@ -85,6 +85,11 @@
                 "static": {"$ref": "#/definitions/static"},
                 "toOne": {"$ref": "#/definitions/toOne"},
                 "toMany": {"$ref": "#/definitions/toManyRecords"},
+                "preserveIdentity": {
+                    "type": "boolean",
+                    "default": False,
+                    "description": "If true, batch edit should update the matched record in place rather than cloning it.",
+                },
             },
             "required": ["wbcols", "static", "toOne", "toMany"],
             "additionalProperties": False,
@@ -101,6 +106,11 @@
                 "static": {"$ref": "#/definitions/static"},
                 "toOne": {"$ref": "#/definitions/toOne"},
                 "toMany": {"$ref": "#/definitions/toManyRecords"},
+                "preserveIdentity": {
+                    "type": "boolean",
+                    "default": False,
+                    "description": "If true, batch edit should update the matched record in place rather than cloning it.",
+                },
             },
             # not making tomany required, to not choke on legacy upload plans
             "required": ["wbcols", "static", "toOne"],
@@ -363,6 +373,7 @@ def rel_table(key: str) -> Table:
         ),
         wbcols={k: parse_column_options(v) for k, v in to_parse["wbcols"].items()},
         static=to_parse["static"],
+        preserveIdentity=to_parse.get("preserveIdentity", False),
         toOne={
             key: parse_uploadable(rel_table(key), to_one)
             for key, to_one in to_parse["toOne"].items()

specifyweb/backend/workbench/upload/upload_table.py

@@ -68,6 +68,7 @@ class UploadTable(NamedTuple):
     toMany: dict[str, list[Uploadable]]
 
     overrideScope: dict[Literal["collection"], int | None] | None = None
+    preserveIdentity: bool = False
 
     def apply_scoping(
         self,
@@ -93,6 +94,8 @@ def _to_json(self) -> dict:
         result = dict(
             wbcols={k: v.to_json() for k, v in self.wbcols.items()}, static=self.static
         )
+        if self.preserveIdentity:
+            result["preserveIdentity"] = True
         result["toOne"] = {
             key: uploadable.to_json() for key, uploadable in self.toOne.items()
         }
@@ -128,6 +131,7 @@ class ScopedUploadTable(NamedTuple):
     static: dict[str, Any]
     toOne: dict[str, ScopedUploadable]
     toMany: dict[str, list["ScopedUploadable"]]  # type: ignore
+    preserveIdentity: bool
     scopingAttrs: dict[str, int]
     disambiguation: int | None
     to_one_fields: dict[str, list[str]]  # TODO: Consider making this a payload..
@@ -257,6 +261,7 @@ def bind(
             parsedFields=parsedFields,
             toOne=toOne,
             toMany=toMany,
+            preserveIdentity=self.preserveIdentity,
             uploadingAgentId=uploadingAgentId,
             auditor=auditor,
             cache=cache,
@@ -326,6 +331,7 @@ class BoundUploadTable(NamedTuple):
     parsedFields: list[ParseResult]
     toOne: dict[str, BoundUploadable]
     toMany: dict[str, list[BoundUploadable]]
+    preserveIdentity: bool
     scopingAttrs: dict[str, int]
     disambiguation: int | None
     uploadingAgentId: int | None
@@ -982,11 +988,21 @@ def _process_to_ones(self) -> dict[str, UploadResult]:
             field_name: (
                 to_one_def.save_row(force=(not self.auditor.props.allow_delete_dependents))
                 if to_one_def.is_one_to_one()
-                else to_one_def.process_row()
+                else (
+                    to_one_def.save_row(force=True)
+                    if self._should_update_to_one_in_place(to_one_def)
+                    else to_one_def.process_row()
+                )
             )
             for field_name, to_one_def in Func.sort_by_key(self.toOne)
         }
 
+    def _should_update_to_one_in_place(self, to_one_def) -> bool:
+        return (
+            getattr(to_one_def, "preserveIdentity", False)
+            and isinstance(getattr(to_one_def, "current_id", None), int)
+        )
+
     def _do_upload(
         self, model, to_one_results: dict[str, UploadResult], info: ReportInfo
     ) -> UploadResult: