Refactor refobj

sbol3/document.py

@@ -213,6 +213,20 @@ def _parse_attributes(objects, graph) -> dict[str, Identified]:
                 obj._owned_objects[str_p].append(other)
                 # Record the assigned object as a child
                 child_objects[other_identity] = other
+            elif str_p in obj._referenced_objects:
+                reference = str(o)
+                # A reference may refer to another object
+                # in the current document, or it may be
+                # an external reference to another RDF entity
+                if reference in objects:
+                    other = objects[reference]
+                    obj._referenced_objects[str_p].append(other)
+                    other._references.append(obj)
+                else:
+                    # If an external reference, create a base SBOLObject to represent it
+                    stub = SBOLObject(reference)
+                    obj._referenced_objects[str_p].append(stub)
+                    stub._references.append(obj)  # Add to reference counter
             elif str_p == RDF_TYPE:
                 # Handle rdf:type specially because the main type(s)
                 # will already be in the list from the build_object
@@ -358,7 +372,27 @@ def _add(self, obj: TopLevel) -> TopLevel:
         # in the TopLevel being added
         def assign_document(x: Identified):
             x.document = self
+
         obj.traverse(assign_document)
+
+        # Update any external references from this object,
+        # which may be resolved upon adding it to the Document.
+        # Stub SBOLObjects will be replaced with the actual
+        # referenced object
+        for property_id, ref_objects in obj._referenced_objects.items():
+            updated = []
+            for ref_obj in ref_objects:
+                resolved_reference = self.find(ref_obj.identity)
+                if resolved_reference:
+                    updated.append(resolved_reference)
+                else:
+                    updated.append(ref_obj)
+            obj._referenced_objects[property_id] = updated
+
+        # Update any external references to this object
+        # replacing stub SBOLObjects with this one
+        self._resolve_references(obj)
+
         return obj
 
     def _add_all(self, objects: pytyping.Sequence[TopLevel]) -> pytyping.Sequence[TopLevel]:
@@ -422,6 +456,11 @@ def find(self, search_string: str) -> Optional[Identified]:
                 return obj
         return self._find_in_objects(search_string)
 
+    def _resolve_references(self, new_obj):
+        """Update all unresolved references to this object, replacing stub SBOLObject with this one."""
+        for updated in self.objects:
+            updated._resolve_references(new_obj)
+
     def join_lines(self, lines: List[Union[bytes, str]]) -> Union[bytes, str]:
         """Join lines for either bytes or strings. Joins a list of lines
         together whether they are bytes or strings. Returns a bytes if the input was
@@ -679,6 +718,11 @@ def remove(self, objects: Iterable[TopLevel]):
         # Now do the removal of each top level object and all of its children
         for obj in objects_to_remove:
             obj.remove_from_document()
+            # If the removed object is referenced anywhere,
+            # leave a stub
+            stub_obj = SBOLObject(obj.identity)
+            assert stub_obj.identity == obj.identity
+            self._resolve_references(stub_obj)
 
     def remove_object(self, top_level: TopLevel):
         """Removes the given TopLevel from this document. No referential

sbol3/identified.py

@@ -269,6 +269,12 @@ def serialize(self, graph: rdflib.Graph):
             rdf_prop = rdflib.URIRef(prop)
             for item in items:
                 graph.add((identity, rdf_prop, item))
+        for prop, items in self._referenced_objects.items():
+            if not items:
+                continue
+            rdf_prop = rdflib.URIRef(prop)
+            for item in items:
+                graph.add((identity, rdf_prop, rdflib.URIRef(item.identity)))
         for prop, items in self._owned_objects.items():
             if not items:
                 continue

sbol3/object.py

@@ -14,10 +14,12 @@ class SBOLObject:
     def __init__(self, name: str) -> None:
         self._properties = defaultdict(list)
         self._owned_objects = defaultdict(list)
+        self._referenced_objects = defaultdict(list)
         # Does this need to be a property? It does not get serialized to the RDF file.
         # Could it be an attribute that gets composed on the fly? Keep it simple for
         # now, and change to a property in the future if needed.
         self._identity = SBOLObject._make_identity(name)
+        self._references = []
 
     def __setattr__(self, name, value):
         try:
@@ -36,6 +38,14 @@ def __getattribute__(self, name):
             result = result.get()
         return result
 
+    def __eq__(self, other):
+        if type(other) is str:
+            return self.identity == other
+        return super().__eq__(other)
+
+    def __hash__(self):
+        return hash(self.identity)
+
     @staticmethod
     def _is_url(name: str) -> bool:
         parsed = urlparse(name)
@@ -96,7 +106,11 @@ def find(self, search_string: str) -> Optional['SBOLObject']:
                     return result
         return None
 
-    def copy(self, target_doc=None, target_namespace=None):
+    def _resolve_references(self, new_obj):
+        resolve_references = make_resolve_references_traverser(new_obj)
+        self.traverse(resolve_references)
+
+    def copy(self, target_doc=None, target_namespace=None):  # pylint: disable=R0912
 
         # Delete this method in v1.1
         warnings.warn('Use sbol3.copy() instead', DeprecationWarning)
@@ -147,8 +161,28 @@ def copy(self, target_doc=None, target_namespace=None):
                 new_obj._owned_objects[property_uri].append(o_copy)
                 o_copy.parent = self
 
+        # After we have copied all the owned objects, copy the referenced objects
+        # and attempt to resolve the references
+        if target_doc:
+            for property_uri, object_store in self._referenced_objects.items():
+                for o in object_store:
+                    referenced_obj = target_doc.find(o.identity)
+                    if referenced_obj:
+                        new_obj._referenced_objects[property_uri].append(referenced_obj)
+                    else:
+                        new_obj._referenced_objects[property_uri].append(SBOLObject(o.identity))
+        else:
+            # If the copy does not belong to a Document, then treat all references
+            # like external references
+            for property_uri, object_store in self._referenced_objects.items():
+                for o in object_store:
+                    new_obj._referenced_objects[property_uri].append(SBOLObject(o.identity))
+
         return new_obj
 
+    def lookup(self):
+        return self
+
 
 def replace_namespace(old_uri, target_namespace, rdf_type):
 
@@ -158,6 +192,25 @@ def replace_namespace(old_uri, target_namespace, rdf_type):
     raise NotImplementedError()
 
 
+def make_resolve_references_traverser(new_obj) -> Callable:
+    # Return a callback for traversing documents and updating
+    # any references to new_obj
+
+    def resolve_references(x):
+        for property_id, references in x._referenced_objects.items():
+            needs_updating = False
+            ref_obj = None
+            for ref_obj in references:
+                if ref_obj.identity == new_obj.identity:
+                    needs_updating = True
+                    break
+            if needs_updating:
+                references.remove(ref_obj)
+                references.append(new_obj)
+
+    return resolve_references
+
+
 # Global store for builder methods. Custom SBOL classes
 # register their builders in this store
 BUILDER_REGISTER: Dict[str, Callable[[str, str], SBOLObject]] = {}

sbol3/refobj_property.py

@@ -22,26 +22,45 @@ def lookup(self):
 
 class ReferencedObjectMixin:
 
+    def _storage(self) -> Dict[str, list]:
+        return self.property_owner._referenced_objects
+
+    def _clear(self):
+        for obj in self._storage()[self.property_uri]:
+            obj._references = []
+
     def to_user(self, value: Any) -> str:
-        result = ReferencedURI(str(value))
         if hasattr(self, 'property_owner'):
             parent = self.property_owner
-            result.parent = parent
-        return result
-
-    @staticmethod
-    def from_user(value: Any) -> rdflib.URIRef:
-        if isinstance(value, SBOLObject):
-            # see https://github.com/SynBioDex/pySBOL3/issues/357
-            if value.identity is None:
-                # The SBOLObject has an uninitialized identity
-                msg = f'Cannot set reference to {value}.'
-                msg += ' Object identity is uninitialized.'
-                raise ValueError(msg)
-            value = value.identity
-        if not isinstance(value, str):
-            raise TypeError(f'Expecting string, got {type(value)}')
-        return rdflib.URIRef(value)
+            value.parent = parent
+        # Should we check to see if value has a document as well?
+        return value
+
+    def from_user(self, value: Any) -> rdflib.URIRef:
+        # TODO: what is value is empty string?
+        if type(value) is str:
+            if self.property_owner.document:
+                referenced_obj = self.property_owner.document.find(value)
+                if referenced_obj is not None:
+                    # Keep track of this reference to the object
+                    if self not in referenced_obj._references:
+                        referenced_obj._references += [self.property_owner]
+                    return referenced_obj
+            # The given URI refers to an object not currently in this
+            # Document, so create a stub
+            # TODO: warn user referenced object is not in document
+            value = SBOLObject(value)
+
+        if not isinstance(value, SBOLObject):
+            raise TypeError('Cannot set property, the value must be str or instance of SBOLObect')
+        if value.identity is None:
+            # The SBOLObject has an uninitialized identity
+            msg = f'Cannot set reference to {value}.'
+            msg += ' Object identity is uninitialized.'
+            raise ValueError(msg)
+        # Keep track of this reference to the object
+        value._references += [self.property_owner]
+        return value
 
     def maybe_add_to_document(self, value: Any) -> None:
         # if not isinstance(value, TopLevel):
@@ -67,6 +86,8 @@ def __init__(self, property_owner: Any, property_uri: str,
             self.set(initial_value)
 
     def set(self, value: Any) -> None:
+        for o in self._storage()[self.property_uri]:
+            o._references.remove(self.property_owner)
         super().set(value)
         # See bug 184 - don't add to document
         # self.maybe_add_to_document(value)
@@ -87,6 +108,25 @@ def __init__(self, property_owner: Any, property_uri: str,
                 initial_value = [initial_value]
             self.set(initial_value)
 
+    def __setitem__(self, key: Union[int, slice], value: Any) -> None:
+        replaced_obj = self._storage()[self.property_uri].__getitem__(key)
+        replaced_obj._references.remove(self.property_owner)
+        super().__setitem__(key, value)
+
+    def __delitem__(self, key: Union[int, slice]) -> None:
+        replaced_item = self._storage()[self.property_uri][key]
+        replaced_item._references.remove(self.property_owner)
+        super().__delitem__(key)
+
+    def set(self, value: Any) -> None:
+        # If the current value of the property
+        # is identical to the value being set, do nothing.
+        if value == self._storage()[self.property_uri]:
+            return
+        for o in self._storage()[self.property_uri]:
+            o._references.remove(self.property_owner)
+        super().set(value)
+
     # See bug 184 - don't add to document
     # def item_added(self, item: Any) -> None:
     #     self.maybe_add_to_document(item)

sbol3/toplevel.py

@@ -140,16 +140,34 @@ def set_identity(self, new_identity: str) -> Any:
         self._display_id = self._extract_display_id(self._identity)
 
     def clone(self, new_identity: str = None) -> 'TopLevel':
-        obj = copy.deepcopy(self)
-        identity_map = {self.identity: obj}
+        clone = copy.deepcopy(self)
+        identity_map = {self.identity: clone}
         # Set identity of new object
         if new_identity is not None:
-            obj.set_identity(new_identity)
+            clone.set_identity(new_identity)
         # Drop the document pointer
-        obj.document = None
+        clone.document = None
 
-        obj.update_all_dependents(identity_map)
-        return obj
+        clone.update_all_dependents(identity_map)
+
+        # Replace references with stub objects, because
+        # the cloned object is not associated with a
+        # Document, resulting in external references
+        def reset_references(x):
+            for property_id, ref_objects in x._referenced_objects.items():
+                updated_objects = []
+                for o in ref_objects:
+                    if clone.find(o.identity):
+                        updated_objects.append(o)
+                    else:
+                        stub = SBOLObject(o.identity)
+                        stub._references = o._references.copy()
+                        updated_objects.append(stub)
+                x._referenced_objects[property_id] = updated_objects
+
+        clone.traverse(reset_references)
+        clone._references = [SBOLObject(r.identity) for r in self._references]
+        return clone
 
     def update_all_dependents(self, identity_map: dict[str, Identified]) -> Any:
         """Update all dependent objects based on the provided identity map.

test/test_collection.py

@@ -35,14 +35,14 @@ def test_member_property(self):
         sbol3.set_namespace('https://github.com/synbiodex/pysbol3')
         self.assertTrue(hasattr(sbol3, 'SBOL_MEMBER'))
         collection = sbol3.Collection('collection1')
-        self.assertIn(sbol3.SBOL_MEMBER, collection._properties)
+        self.assertIn(sbol3.SBOL_MEMBER, collection._referenced_objects)
         self.assertNotIn(sbol3.SBOL_ORIENTATION, collection._properties)
         uris = ['https://github.com/synbiodex/pysbol3/thing1',
                 'https://github.com/synbiodex/pysbol3/thing2']
         collection.members = uris
-        self.assertIn(sbol3.SBOL_MEMBER, collection._properties)
-        self.assertNotIn(sbol3.SBOL_ORIENTATION, collection._properties)
-        self.assertEqual(uris, collection.members)
+        self.assertIn(sbol3.SBOL_MEMBER, collection._referenced_objects)
+        self.assertNotIn(sbol3.SBOL_ORIENTATION, collection._referenced_objects)
+        self.assertListEqual(uris, [m.identity for m in collection.members])  # pylint: disable=E1101
 
     # Namespace testing
     def test_namespace_deduced(self):

test/test_component.py

@@ -94,7 +94,8 @@ def test_cloning_with_references(self):
         c2 = c1.clone(new_identity)
         self.assertEqual(posixpath.join(sbol3.get_namespace(), new_identity),
                          c2.identity)
-        self.assertListEqual(list(c1.sequences), list(c2.sequences))
+        self.assertListEqual([s.identity for s in c1.sequences],
+                             [s.identity for s in c2.sequences])
 
     def test_cloning_with_children(self):
         # This test does not use `sbol3.set_namespace` as the other
@@ -129,7 +130,7 @@ def test_cloning_with_children(self):
         self.assertIsInstance(es2, sbol3.EntireSequence)
         self.assertNotEqual(es1.identity, es2.identity)
         self.assertTrue(es2.identity.startswith(c2.identity))
-        self.assertEqual(es1.sequence, es2.sequence)
+        self.assertEqual(es1.sequence.identity, es2.sequence.identity)
         self.assertIsNone(es2.document)
 
     def test_cloning_references(self):
@@ -159,15 +160,15 @@ def test_cloning_references(self):
             self.assertIsInstance(s_clone, sbol3.ComponentReference)
             self.assertTrue(s_clone.identity.startswith(toggle_clone.identity))
             self.assertNotEqual(s.identity, s_clone.identity)
-            self.assertEqual(s.refers_to, s_clone.refers_to)
+            self.assertEqual(s.refers_to.identity, s_clone.refers_to.identity)
             o = c.object.lookup()
             self.assertIsInstance(o, sbol3.ComponentReference)
             self.assertTrue(o.identity.startswith(toggle.identity))
             o_clone = c_clone.object.lookup()
             self.assertIsInstance(o_clone, sbol3.ComponentReference)
             self.assertTrue(o_clone.identity.startswith(toggle_clone.identity))
             self.assertNotEqual(o.identity, o_clone.identity)
-            self.assertEqual(o.refers_to, o_clone.refers_to)
+            self.assertEqual(o.refers_to.identity, o_clone.refers_to.identity)
 
     def test_measures_initial_value(self):
         # See https://github.com/SynBioDex/pySBOL3/issues/301

test/test_implementation.py

@@ -41,7 +41,7 @@ def test_read_from_file(self):
         self.assertIsInstance(implementation, sbol3.Implementation)
         tetr_uri = 'https://sbolstandard.org/examples/TetR_protein'
         built = tetr_uri
-        self.assertCountEqual(built, implementation.built)
+        self.assertCountEqual(built, implementation.built.identity)
 
 
 if __name__ == '__main__':