Only create remote to-one resources after object is created

specifyweb/businessrules/tests/test_cojo.py

@@ -1,4 +1,5 @@
-from specifyweb.specify.models import Collectionobjectgroup, Collectionobjectgroupjoin, Collectionobjectgrouptype, Picklist, Picklistitem
+from specifyweb.specify.models import Collectionobjectgroup, Collectionobjectgroupjoin, Collectionobjectgrouptype
+from specifyweb.businessrules.exceptions import BusinessRuleException
 from specifyweb.specify.tests.test_api import DefaultsSetup
 
 class CoJoTest(DefaultsSetup):
@@ -58,3 +59,21 @@ def test_cojo_rules_enforcement(self):
         self.assertTrue(cojo_2.issubstrate)
         self.assertFalse(cojo_3.isprimary)
         self.assertFalse(cojo_3.issubstrate)
+
+        with self.assertRaises(BusinessRuleException): 
+            Collectionobjectgroupjoin.objects.create(
+                isprimary=False,
+                issubstrate=False,
+                parentcog=cog_1,
+                childcog=cog_4,
+                childco=self.collectionobjects[0]
+            )
+
+        with self.assertRaises(BusinessRuleException): 
+            Collectionobjectgroupjoin.objects.create(
+                isprimary=False,
+                issubstrate=False,
+                parentcog=cog_1,
+                childcog=None,
+                childco=None
+            )

specifyweb/specify/api.py

@@ -7,7 +7,7 @@
 import logging
 import re
 from typing import Any, Dict, List, Optional, Tuple, Iterable, Union, \
-    Callable, TypedDict, cast
+    Callable, TypedDict, Literal, cast
 
 from urllib.parse import urlencode
 
@@ -513,7 +513,7 @@ def create_obj(collection, agent, model, data: Dict[str, Any], parent_obj=None,
 
     data = cleanData(model, data, parent_relationship)
     obj = model()
-    _, remote_to_ones, _ = handle_fk_fields(collection, agent, obj, data)
+    _, _, handle_remote_to_ones = handle_fk_fields(collection, agent, obj, data)
     set_fields_from_data(obj, data)
     set_field_if_exists(obj, 'createdbyagent', agent)
     set_field_if_exists(obj, 'collectionmemberid', collection.id)
@@ -526,9 +526,7 @@ def create_obj(collection, agent, model, data: Dict[str, Any], parent_obj=None,
         check_table_permissions(collection, agent, obj, "create")
         auditlog.insert(obj, agent, parent_obj)
 
-    for (field, related) in remote_to_ones:
-        setattr(related, field.name, obj)
-        related.save()
+    handle_remote_to_ones(obj)
     handle_to_many(collection, agent, obj, data)
     return obj
 
@@ -605,8 +603,7 @@ def reorder_fields_for_embedding(cls, data: Dict[str, Any]) -> Iterable[Tuple[st
     for key in data.keys() - {put_first}:
         yield (key, data[key])
 
-
-def handle_fk_fields(collection, agent, obj, data: Dict[str, Any]) -> Tuple[List, List, List[FieldChangeInfo]]:
+def handle_fk_fields(collection, agent, obj, data: Dict[str, Any]) -> Tuple[List, List[FieldChangeInfo], Callable[[Any], None]]:
     """Where 'obj' is a Django model instance and 'data' is a dict,
     set foreign key fields in the object from the provided data.
     """
@@ -622,54 +619,122 @@ def handle_fk_fields(collection, agent, obj, data: Dict[str, Any]) -> Tuple[List
         field = obj._meta.get_field(field_name)
         if not field.many_to_one and not field.one_to_one: continue
 
-        old_related = get_related_or_none(obj, field_name)
-        dependent = is_dependent_field(obj, field_name)
-        old_related_id = None if old_related is None else old_related.id
-        new_related_id = None
-
-        if val is None:
-            setattr(obj, field_name, None)
-            rel_obj = None
-            if dependent and old_related:
-                dependents_to_delete.append(old_related)
-
-        elif isinstance(val, field.related_model):
-            # The related value was patched into the data by a parent object.
-            setattr(obj, field_name, val)
-            rel_obj = val
-            new_related_id = val.id
-
-        elif isinstance(val, str):
-            # The related object is given by a URI reference.
-            assert not dependent, "didn't get inline data for dependent field %s in %s: %r" % (field_name, obj, val)
-            fk_model, fk_id = strict_uri_to_model(val, field.related_model.__name__)
-            rel_obj = get_object_or_404(fk_model, id=fk_id)
-            setattr(obj, field_name, rel_obj)
-            new_related_id = fk_id
-
-        elif hasattr(val, 'items'):  # i.e. it's a dict of some sort
-            # The related object is represented by a nested dict of data.
-            rel_model = field.related_model
-            datamodel_field = obj.specify_model.get_relationship(field_name)
-            rel_obj = update_or_create_resource(collection, agent, rel_model, val, obj if dependent else None, datamodel_field)
-
-            setattr(obj, field_name, rel_obj)
-            if dependent and old_related and old_related.id != rel_obj.id:
-                dependents_to_delete.append(old_related)
-            new_related_id = rel_obj.id
-            data[field_name] = _obj_to_data(rel_obj, read_checker)
+        if field.concrete: 
+            some_remote_to_ones = []
+            extra_data, some_dependents_to_delete, some_dirty = _handle_fk_field(collection, agent, obj, field, val, read_checker)
         else:
-            raise Exception(f'bad foreign key field in data: {field_name}')
-
-        if str(old_related_id) != str(new_related_id):
-            dirty.append({'field_name': field_name, 'old_value': old_related_id, 'new_value': new_related_id})
+            extra_data = dict()
+            some_dependents_to_delete, some_dirty, some_remote_to_ones = _handle_remote_fk_field(obj, field, val, read_checker)
 
-        # If the field is remote, the obj must be set on the remote side once 
-        # the obj exists
-        if not field.concrete and rel_obj is not None: 
-                remote_to_ones.append((field.remote_field, rel_obj))
+        data.update(extra_data)
+        dirty += some_dirty
+        dependents_to_delete += some_dependents_to_delete
+        remote_to_ones += some_remote_to_ones
+
+    def _set_remote_to_ones(created_obj): 
+        for field, related_data, is_dependent in remote_to_ones: 
+            related_data[field.name] = created_obj
+            rel_obj = update_or_create_resource(collection, agent, field.model, related_data, created_obj if is_dependent else None)
+            setattr(obj, field.remote_field.name, rel_obj)
+
+    return dependents_to_delete, dirty, _set_remote_to_ones
+
+def _handle_fk_field(collection, agent, obj, field, value, read_checker): 
+    field_name = field.name
+    old_related = get_related_or_none(obj, field_name)
+    dependent = is_dependent_field(obj, field_name)
+    old_related_id = None if old_related is None else old_related.id
+    new_related_id = None
+
+    dependents_to_delete = []
+    dirty: List[FieldChangeInfo] = []
+    data = dict()
+
+    if value is None:
+        setattr(obj, field_name, None)
+        rel_obj = None
+        if dependent and old_related:
+            dependents_to_delete.append(old_related)
+
+    elif isinstance(value, field.related_model):
+        # The related value was patched into the data by a parent object.
+        setattr(obj, field_name, value)
+        rel_obj = value
+        new_related_id = value.id
+
+    elif isinstance(value, str):
+        # The related object is given by a URI reference.
+        assert not dependent, "didn't get inline data for dependent field %s in %s: %r" % (field_name, obj, value)
+        fk_model, fk_id = strict_uri_to_model(value, field.related_model.__name__)
+        rel_obj = get_object_or_404(fk_model, id=fk_id)
+        setattr(obj, field_name, rel_obj)
+        new_related_id = fk_id
+
+    elif hasattr(value, 'items'):  # i.e. it's a dict of some sort
+        # The related object is represented by a nested dict of data.
+        rel_model = field.related_model
+        datamodel_field = obj.specify_model.get_relationship(field_name)
+        rel_obj = update_or_create_resource(collection, agent, rel_model, value, obj if dependent else None, datamodel_field)
+
+        setattr(obj, field_name, rel_obj)
+        if dependent and old_related and old_related.id != rel_obj.id:
+            dependents_to_delete.append(old_related)
+        new_related_id = rel_obj.id
+        data[field_name] = _obj_to_data(rel_obj, read_checker)
+    else:
+        raise Exception(f'bad foreign key field in data: {field_name}')
+
+    if str(old_related_id) != str(new_related_id):
+        dirty.append({'field_name': field_name, 'old_value': old_related_id, 'new_value': new_related_id})
+    return data, dependents_to_delete, dirty
+
+def _handle_remote_fk_field(obj, field, value, read_checker): 
+    field_name = field.name
+    old_related = get_related_or_none(obj, field_name)
+    dependent = is_dependent_field(obj, field_name)
+    old_related_id = None if old_related is None else old_related.id
+    new_related_id = None
 
-    return dependents_to_delete, remote_to_ones, dirty
+    remote_to_ones = []
+    dependents_to_delete = []
+    dirty: List[FieldChangeInfo] = []
+
+    if value is None:
+        rel_data = None
+        setattr(obj, field_name, None)
+        if dependent and old_related:
+            dependents_to_delete.append(old_related)
+
+    elif isinstance(value, field.related_model):
+        # The related value was patched into the data by a parent object.
+        setattr(obj, field_name, value)
+        rel_data = _obj_to_data(value, read_checker)
+        new_related_id = value.id
+
+    elif isinstance(value, str):
+        # The related object is given by a URI reference.
+        assert not dependent, "didn't get inline data for dependent field %s in %s: %r" % (field_name, obj, value)
+        fk_model, fk_id = strict_uri_to_model(value, field.related_model.__name__)
+        rel_obj = get_object_or_404(fk_model, id=fk_id)
+        setattr(obj, field_name, rel_obj)
+        rel_data = _obj_to_data(rel_obj, read_checker)
+        new_related_id = rel_obj.pk
+
+    elif hasattr(value, 'items'):  # i.e. it's a dict of some sort
+        # The related object is represented by a nested dict of data.
+        rel_data = value
+        new_related_id = value.get("id", None)
+        if dependent and old_related and new_related_id and old_related.id != new_related_id:
+            dependents_to_delete.append(old_related)
+    else:
+        raise Exception(f'bad foreign key field in data: {field_name}')
+
+    if str(old_related_id) != str(new_related_id):
+        dirty.append({'field_name': field_name, 'old_value': old_related_id, 'new_value': new_related_id})
+
+    if not rel_data is None: 
+        remote_to_ones.append((field.remote_field, rel_data, dependent))
+    return dependents_to_delete, dirty, remote_to_ones
 
 def handle_to_many(collection, agent, obj, data: Dict[str, Any]) -> None:
     """For every key in the dict 'data' which is a *-to-many field in the
@@ -705,7 +770,7 @@ def handle_to_many(collection, agent, obj, data: Dict[str, Any]) -> None:
 def _handle_dependent_to_many(collection, agent, obj, field, value):
     if not isinstance(value, list): 
         assert isinstance(value, list), "didn't get inline data for dependent field %s in %s: %r" % (field.name, obj, value)
-        
+
     rel_model = field.related_model
     ids = [] # Ids not in this list will be deleted at the end.
 
@@ -831,7 +896,7 @@ def update_obj(collection, agent, name: str, id, version, data: Dict[str, Any],
     check_table_permissions(collection, agent, obj, "update")
 
     data = cleanData(obj.__class__, data, parent_relationship)
-    dependents_to_delete, remote_to_ones, fk_dirty = handle_fk_fields(collection, agent, obj, data)
+    dependents_to_delete, fk_dirty, handle_remote_to_ones = handle_fk_fields(collection, agent, obj, data)
     dirty = fk_dirty + set_fields_from_data(obj, data)
 
     check_field_permissions(collection, agent, obj, [d['field_name'] for d in dirty], "update")
@@ -849,9 +914,7 @@ def update_obj(collection, agent, name: str, id, version, data: Dict[str, Any],
     auditlog.update(obj, agent, parent_obj, dirty)
     for dep in dependents_to_delete:
         delete_obj(dep, parent_obj=obj, collection=collection, agent=agent)
-    for (field, related) in remote_to_ones:
-        setattr(related, field.name, obj)
-        related.save()
+    handle_remote_to_ones(obj)
     handle_to_many(collection, agent, obj, data)
     return obj
 

specifyweb/specify/tests/test_api.py

@@ -690,6 +690,92 @@ def test_skipping_redundant_resources(self):
         accession = api.create_obj(self.collection, self.agent, 'Accession', accession_data)
         self.assertFalse(models.Accession.objects.filter(accessionnumber=redundant_accession_number).exists())
         self.assertFalse(models.Collectionobject.objects.filter(catalognumber=redundant_catalog_number).exists())
+
+class InlineApiRemoteToOneTests(ApiTests): 
+    def setUp(self): 
+        super(InlineApiRemoteToOneTests, self).setUp()
+        cog_type_picklist = Picklist.objects.create(
+            name=SYSTEM_COGTYPES_PICKLIST,
+            issystem=True,
+            type=0,
+            readonly=True,
+            collection=self.collection
+        )
+        Picklistitem.objects.create(
+            title='Discrete',
+            value='Discrete',
+            picklist=cog_type_picklist
+        )
+        self.cogtype = models.Collectionobjectgrouptype.objects.create(
+            name="Discrete", type="Discrete", collection=self.collection
+        )
+        self.cog_parent = models.Collectionobjectgroup.objects.create(
+            name="Parent",
+            cogtype=self.cogtype,
+            collection=self.collection,
+        )
+
+    def test_setting_remote_to_one_from_new(self):
+        co_data = {
+            "catalognumber": f'num-{len(self.collectionobjects)}',
+            "cojo": {
+                "isPrimary": True,
+                "isSubstrate": False,
+                "parentCog": api.uri_for_model("Collectionobjectgroup", self.cog_parent.id)
+            },
+            'collection': api.uri_for_model('Collection', self.collection.id),
+        }
+        co = api.create_obj(self.collection, self.agent, "Collectionobject", co_data)
+        cojo = models.Collectionobjectgroupjoin.objects.get(parentcog_id=self.cog_parent.id, childco=co)
+        self.assertEqual(co.cojo, cojo)
+
+    def test_setting_remote_to_one_from_existing(self): 
+        existing_co = self.collectionobjects[0]
+        co_data = {
+            **api.obj_to_data(existing_co),
+            "cojo": {
+                "isPrimary": True,
+                "isSubstrate": False,
+                "parentCog": api.uri_for_model("Collectionobjectgroup", self.cog_parent.id)
+            },
+        }
+        co = api.update_obj(self.collection, self.agent, "Collectionobject", existing_co.id, existing_co.version, co_data)
+        cojo = models.Collectionobjectgroupjoin.objects.get(parentcog_id=self.cog_parent.id, childco=co)
+        self.assertEqual(co.cojo, cojo)
+
+    def test_creating_independent_from_remote_one_to_one(self):
+        new_parent_name = "ParentTwo"
+        co_data = {
+            "catalognumber": f'num-{len(self.collectionobjects)}',
+            "cojo": {
+                "isPrimary": True,
+                "isSubstrate": False,
+                "parentCog": {
+                    "name": new_parent_name,
+                    "cogtype": api.uri_for_model("Collectionobjectgrouptype", self.cogtype.id),
+                    'collection': api.uri_for_model('Collection', self.collection.id)
+                }
+            },
+            'collection': api.uri_for_model('Collection', self.collection.id),
+        }
+        co = api.create_obj(self.collection, self.agent, "Collectionobject", co_data)
+        cojo = models.Collectionobjectgroupjoin.objects.get(parentcog__name=new_parent_name, childco=co)
+
+        self.assertEqual(co.cojo, cojo)
+        self.assertEqual(co.cojo.parentcog.name, new_parent_name)
+
+    def test_unsetting_dependent_remote_one_to_one(self): 
+        existing_co = self.collectionobjects[1]
+        cojo = models.Collectionobjectgroupjoin.objects.create(isprimary=False, parentcog=self.cog_parent,childco=existing_co)
+        existing_co.refresh_from_db()
+        self.assertEqual(existing_co.cojo, cojo)
+        co_data = {
+            **api.obj_to_data(existing_co),
+            "cojo": None,
+        }
+        co = api.update_obj(self.collection, self.agent, "Collectionobject", existing_co.id, existing_co.version, co_data)
+        self.assertIsNone(api.get_related_or_none(co, "cojo"))
+        self.assertFalse(models.Collectionobjectgroupjoin.objects.filter(childco_id=co.id).exists())
 
     # version control on inlined resources should be tested