From ca31bac59cfd0c4ac82450bc6dd1039c4aa9b903 Mon Sep 17 00:00:00 2001 From: Andrew Davison Date: Mon, 14 Sep 2026 15:56:05 +0200 Subject: [PATCH] Look for existing instances in all spaces when saving (fixes #136) save() restricted its existence query to the space it was about to write to, so a locally-constructed object whose counterpart already lives in a different space was not recognised, and a duplicate was created. This happened most often with recursive=True, where new child objects inherit the parent's target space. The restriction was intended as an optimization, but benchmarking against the pre-production KG showed no measurable benefit: Person queries took the same time whether or not they were restricted to a space, and File queries timed out after ~50 s even when restricted to a space containing no files. save() now runs a single existence query across all spaces, which also returns the space of each matching instance. If one of the matches is in the target space, that instance is used, and matches in other spaces are not treated as duplicates. If the only match is in another space, it is used and updated in the space where it lives, with a warning, rather than a duplicate being created in the requested space. Multiple matches, none of them in the target space, raise an exception unless ignore_duplicates=True. This resolves the long-standing TODO about existing objects in a different space. To let save() choose between matches, the query part of exists() is split into private methods (_exists_without_query, _query_matching_instances, _check_for_duplicates, _use_matching_instance), shared by exists() and save(). The signature and behaviour of exists(), including in_spaces, are unchanged. When exists() binds to a match found by the query, the object's space is set to the space of that instance. Existence queries request EXISTENCE_QUERY_SIZE (2) results. If the response reports a larger total, all matches are retrieved in a second query, so that an instance in the target space is not missed. In exists(), a ConnectionError other than RemoteDisconnected was swallowed, leaving the result undefined and raising a NameError. It is now re-raised. MockKGClient now returns the space of matching instances under the name used by the query API. New tests cover children and top-level objects found in another space, preferring the target space, duplicates in the target space and in other spaces, retrieving all matches, and connection errors. --- fairgraph/kgobject.py | 203 ++++++++++++++++++++++++------------ test/test_openminds_core.py | 189 ++++++++++++++++++++++++++++++++- test/utils.py | 6 +- 3 files changed, 331 insertions(+), 67 deletions(-) diff --git a/fairgraph/kgobject.py b/fairgraph/kgobject.py index e2bad6c4..dc381349 100644 --- a/fairgraph/kgobject.py +++ b/fairgraph/kgobject.py @@ -55,6 +55,10 @@ logger = logging.getLogger("fairgraph") +# Number of results requested by existence queries, which must be more than one, to detect duplicates. +# If more instances match, they are all retrieved in a second query. +EXISTENCE_QUERY_SIZE = 2 + class KGObject(KGNode, Releasable): """ @@ -555,8 +559,31 @@ def diff(self, other): return differences def exists(self, client: KGClient, ignore_duplicates: bool = False, in_spaces: Optional[List[str]] = None) -> bool: - """Check if this object already exists in the KnowledgeGraph""" + """ + Check if this object already exists in the KnowledgeGraph. + + Args: + client: KGClient object that handles the communication with the KG. + ignore_duplicates (bool, optional): Whether to ignore the existence of multiple objects with the same properties + (and consider only the first in the list), or to raise an Exception. Defaults to False. + in_spaces (list of str, optional): If provided, only look for the object in these spaces. + """ + obj_exists = self._exists_without_query(client) + if obj_exists is not None: + return obj_exists + instances = self._query_matching_instances(client, in_spaces=in_spaces) + if not instances: + return False + self._check_for_duplicates(instances, ignore_duplicates) + return self._use_matching_instance(client, instances[0]) + def _exists_without_query(self, client: KGClient) -> Optional[bool]: + """ + Check if this object exists in the KG, where this can be determined without an existence query, + i.e. if the object has an ID, if there is no existence query, or if the object is found in the save cache. + + Returns True or False if this could be determined, None if an existence query is needed. + """ if self.id and self.id.startswith("http"): # Since the KG now allows user-specified IDs we can't assume that the presence of # an id means the object exists @@ -569,70 +596,92 @@ def exists(self, client: KGClient, ignore_duplicates: bool = False, in_spaces: O if obj_exists: self._update_empty_properties(data) # also updates `remote_data` return obj_exists - else: - try: - query_filter = self._build_existence_query() - except CannotBuildExistenceQuery: - return False - if query_filter is None: - # if there's no existence query and no ID, we allow - # duplicate entries - return False - else: - query_cache_key = generate_cache_key(query_filter) - if query_cache_key in save_cache[self.__class__]: - # Because the KnowledgeGraph is only eventually consistent, an instance - # that has just been written to the KG may not appear in the query. - # Therefore we cache the query when creating an instance and - # where exists() returns True - self.id = save_cache[self.__class__][query_cache_key] - cached_obj = object_cache.get(self.id) - if cached_obj and cached_obj.remote_data: - self._raw_remote_data = cached_obj._raw_remote_data - # this also updates `self.remote_data`. It must not be replaced by a - # direct assignment to `self.remote_data`: a property that is empty - # locally but present remotely would then look like a deliberate - # deletion, and be set to null by the next call to save(). - self._update_empty_properties(cached_obj.remote_data) - return True - - query = self.__class__.generate_minimal_query( - client=client, - filters=query_filter, - ) + try: + query_filter = self._build_existence_query() + except CannotBuildExistenceQuery: + return False + if query_filter is None: + # if there's no existence query and no ID, we allow + # duplicate entries + return False + + query_cache_key = generate_cache_key(query_filter) + if query_cache_key in save_cache[self.__class__]: + # Because the KnowledgeGraph is only eventually consistent, an instance + # that has just been written to the KG may not appear in the query. + # Therefore we cache the query when creating an instance and + # where exists() returns True + self.id = save_cache[self.__class__][query_cache_key] + cached_obj = object_cache.get(self.id) + if cached_obj and cached_obj.remote_data: + self._raw_remote_data = cached_obj._raw_remote_data + # this also updates `self.remote_data`. It must not be replaced by a + # direct assignment to `self.remote_data`: a property that is empty + # locally but present remotely would then look like a deliberate + # deletion, and be set to null by the next call to save(). + self._update_empty_properties(cached_obj.remote_data) + return True + return None - try: - instances = client.query( - query=query, size=2, release_status="any", restrict_to_spaces=in_spaces - ).data - except ConnectionError as err: - if "RemoteDisconnected" in str(err): - warn( - f"Timeout when checking for existence of object {self}." - "Returning False, check for possible creation of duplicate instances." - ) - return False + def _query_matching_instances(self, client: KGClient, in_spaces: Optional[List[str]] = None) -> List[JSONdict]: + """ + Run the existence query for this object, and return all matching instances + (their "@id" and space only). - if instances: - if len(instances) > 1 and not ignore_duplicates: - # we might want to consider running a second query with "equals" rather than "contains" - raise Exception( - f"Existence query is not specific enough. Type: {self.__class__.__name__}; filters: {query_filter}" - ) + If the connection is lost while querying, an empty list is returned, with a warning. + """ + query_filter = self._build_existence_query() + query = self.__class__.generate_minimal_query(client=client, filters=query_filter) + try: + response = client.query( + query=query, size=EXISTENCE_QUERY_SIZE, release_status="any", restrict_to_spaces=in_spaces + ) + instances = response.data or [] + if response.total > len(instances): + response = client.query( + query=query, size=response.total, release_status="any", restrict_to_spaces=in_spaces + ) + instances = response.data or [] + except ConnectionError as err: + if "RemoteDisconnected" in str(err): + warn( + f"Timeout when checking for existence of object {self}." + "Returning False, check for possible creation of duplicate instances." + ) + return [] + raise + return instances + + def _check_for_duplicates(self, instances: List[JSONdict], ignore_duplicates: bool): + if len(instances) > 1 and not ignore_duplicates: + # we might want to consider running a second query with "equals" rather than "contains" + raise Exception( + f"Existence query is not specific enough. Type: {self.__class__.__name__}; " + f"filters: {self._build_existence_query()}" + ) - # it seems that sometimes the "query" endpoint returns instances - # which the "instances" endpoint doesn't know about, so here we double check that - # the instance can be found - instance = client.instance_from_full_uri(instances[0]["@id"], release_status="any") - if instance is None: - return False + def _use_matching_instance(self, client: KGClient, match: JSONdict) -> bool: + """ + Identify this object with an instance found by the existence query. - self.id = instance["@id"] - assert isinstance(self.id, str) - save_cache[self.__class__][query_cache_key] = self.id - self._update_empty_properties(instance) # also updates `remote_data` - return bool(instances) + Returns False if the instance could not be retrieved. + """ + # it seems that sometimes the "query" endpoint returns instances + # which the "instances" endpoint doesn't know about, so here we double check that + # the instance can be found + instance = client.instance_from_full_uri(match["@id"], release_status="any") + if instance is None: + return False + + self.id = instance["@id"] + assert isinstance(self.id, str) + # the instance's actual location takes precedence over any space set locally + if "https://schema.hbp.eu/myQuery/space" in match: + self._space = match["https://schema.hbp.eu/myQuery/space"] + save_cache[self.__class__][generate_cache_key(self._build_existence_query())] = self.id + self._update_empty_properties(instance) # also updates `remote_data` + return True def modified_data(self) -> JSONdict: """ @@ -744,7 +793,28 @@ def save( else: space = self.space logger.info(f"Saving a {self.__class__.__name__} in space {space}") - if self.exists(client, ignore_duplicates=ignore_duplicates, in_spaces=[space]): + found = self._exists_without_query(client) + if found is None: + # We look for the object in all spaces, not only the one we are saving to, to avoid creating duplicates, + # but if it exists both in the target space and elsewhere, we use the instance in the target space. + instances = self._query_matching_instances(client) + candidates = [ + instance for instance in instances if instance.get("https://schema.hbp.eu/myQuery/space") == space + ] or instances + if candidates: + self._check_for_duplicates(candidates, ignore_duplicates) + found = self._use_matching_instance(client, candidates[0]) + else: + found = False + if found and self.space is not None and self.space != space: + # An existing instance can only be updated in the space where it lives, + # so we link to it there rather than creating a duplicate in the requested space. + warn( + f"{self.__class__.__name__}(id={self.id}) already exists in space '{self.space}', " + f"so it will be updated there rather than created in space '{space}'" + ) + space = self.space + if found: if not self.allow_update: logger.info(f" - not updating {self.__class__.__name__}(id={self.id}), update not allowed by user") if activity_log: @@ -861,8 +931,8 @@ def save( if activity_log: activity_log.update(item=self, delta=instance_data, space=self.space, entry_type="create") - # not handled yet: if an existing object is in a different space to the one specified here, - # should we move it to the new space, or raise an Exception? + # note: if an existing object is in a different space to the one specified here, + # it is updated in its own space (with a warning), not moved to the new space. if self.id: logger.debug( "Updating cache for object {}. Current state: {}".format( @@ -1086,7 +1156,7 @@ def generate_minimal_query( ) -> Union[Dict[str, Any], None]: """ Generate a minimal KG query definition as a JSON-LD document. - Such a query returns only the @id of any instances that are found. + Such a query returns only the @id, @type and space of any instances that are found. Args: client: KGClient object that handles the communication with the KG. @@ -1106,7 +1176,10 @@ def generate_minimal_query( node_type=cls.type_, label=label, space=None, - properties=[QueryProperty("@type")], + properties=[ + QueryProperty("https://core.kg.ebrains.eu/vocab/meta/space", name="query:space"), + QueryProperty("@type"), + ], ) # second pass, we add filters query.properties.extend(cls.generate_query_filter_properties(normalized_filters)) diff --git a/test/test_openminds_core.py b/test/test_openminds_core.py index e1a09a02..b765fb08 100644 --- a/test/test_openminds_core.py +++ b/test/test_openminds_core.py @@ -8,7 +8,10 @@ import tempfile import urllib.request +from http.client import RemoteDisconnected + import pytest +from requests.exceptions import ConnectionError from openminds import IRI from openminds.base import LinkedNodeEmbedding @@ -16,7 +19,7 @@ from fairgraph.utility import as_list from fairgraph.kgproxy import KGProxy from fairgraph.kgquery import KGQuery -from fairgraph.kgobject import KGObject +from fairgraph.kgobject import KGObject, EXISTENCE_QUERY_SIZE import fairgraph.openminds.core as omcore import fairgraph.openminds.controlled_terms as omterms from fairgraph.utility import ActivityLog, sha1sum, normalize_data @@ -830,6 +833,190 @@ def test_save_new_recursive_mock(mock_client, clear_caches): assert UUID(new_person.affiliations.member_of.uuid) +def _seed_person(mock_client, space, uuid="12345678-90ab-cdef-0123-4567890abcde"): + person_id = f"https://kg.ebrains.eu/api/instances/{uuid}" + mock_client.instances[person_id] = { + "@id": person_id, + "@type": ["https://openminds.om-i.org/types/Person"], + "https://core.kg.ebrains.eu/vocab/meta/space": space, + "https://openminds.om-i.org/props/givenName": "Bilbo", + "https://openminds.om-i.org/props/familyName": "Baggins", + } + return person_id + + +def _spy_on_query(mock_client, monkeypatch): + """Record the `restrict_to_spaces` argument of each query sent to the mock client""" + restrictions = [] + original_query = mock_client.query + + def query(*args, **kwargs): + restrictions.append(kwargs.get("restrict_to_spaces", None)) + return original_query(*args, **kwargs) + + monkeypatch.setattr(mock_client, "query", query) + return restrictions + + +def test_save_recursive_finds_child_in_other_space(mock_client, clear_caches): + """ + A locally-constructed child that already exists in a different space from its parent + should be linked to the existing instance, not duplicated in the parent's space (#136). + """ + person_id = _seed_person(mock_client, "common") + model = omcore.Model( + name="Dummy new model with an existing developer", + developers=omcore.Person(given_name="Bilbo", family_name="Baggins"), + ) + log = ActivityLog() + with pytest.warns(UserWarning, match="already exists in space 'common'.*rather than created in space 'myspace'"): + model.save(mock_client, space="myspace", recursive=True, activity_log=log) + + person_instances = [ + instance + for instance in mock_client.instances.values() + if "https://openminds.om-i.org/types/Person" in instance["@type"] + ] + assert len(person_instances) == 1 + assert model.developers.id == person_id + assert model.developers.space == "common" + person_entries = [entry for entry in log.entries if entry.cls == "Person"] + assert len(person_entries) == 1 + assert person_entries[0].space == "common" + assert person_entries[0].type != "create" + assert model.space == "myspace" + + +def test_save_finds_object_in_other_space(mock_client, clear_caches): + person_id = _seed_person(mock_client, "common") + person = omcore.Person(given_name="Bilbo", family_name="Baggins") + log = ActivityLog() + with pytest.warns(UserWarning, match="already exists in space 'common'"): + person.save(mock_client, space="myspace", recursive=False, activity_log=log) + assert person.id == person_id + assert person.space == "common" + assert len(mock_client.instances) == 1 + assert [(entry.type, entry.space) for entry in log.entries] == [("no-op", "common")] + + +def test_save_prefers_instance_in_target_space(mock_client, clear_caches, monkeypatch, recwarn): + """ + If the object exists both in the target space and elsewhere, the instance in the target space + is used, and the other one is not treated as a duplicate. + """ + _seed_person(mock_client, "common") + target_id = _seed_person(mock_client, "myspace", uuid="23456789-0abc-def0-1234-567890abcdef") + restrictions = _spy_on_query(mock_client, monkeypatch) + person = omcore.Person(given_name="Bilbo", family_name="Baggins") + log = ActivityLog() + person.save(mock_client, space="myspace", recursive=False, activity_log=log) + assert person.id == target_id + assert person.space == "myspace" + assert restrictions == [None] + assert [(entry.type, entry.space) for entry in log.entries] == [("no-op", "myspace")] + assert not [w for w in recwarn if "already exists in space" in str(w.message)] + + +def test_save_new_object_queries_all_spaces_once(mock_client, clear_caches, monkeypatch): + _seed_person(mock_client, "common") + restrictions = _spy_on_query(mock_client, monkeypatch) + person = omcore.Person(given_name="Frodo", family_name="Baggins") + log = ActivityLog() + person.save(mock_client, space="myspace", recursive=False, activity_log=log) + assert restrictions == [None] + assert [(entry.type, entry.space) for entry in log.entries] == [("create", "myspace")] + assert len(mock_client.instances) == 2 + + +def test_save_object_with_local_space_found_in_other_space(mock_client, clear_caches): + person_id = _seed_person(mock_client, "common") + person = omcore.Person(given_name="Bilbo", family_name="Baggins", space="myspace") + log = ActivityLog() + with pytest.warns(UserWarning, match="already exists in space 'common'.*rather than created in space 'myspace'"): + person.save(mock_client, recursive=False, activity_log=log) + assert person.id == person_id + assert person.space == "common" + assert [(entry.type, entry.space) for entry in log.entries] == [("no-op", "common")] + assert len(mock_client.instances) == 1 + + +def test_save_duplicates_in_target_space(mock_client, clear_caches): + _seed_person(mock_client, "common") + _seed_person(mock_client, "myspace", uuid="23456789-0abc-def0-1234-567890abcdef") + _seed_person(mock_client, "myspace", uuid="34567890-abcd-ef01-2345-67890abcdef0") + person = omcore.Person(given_name="Bilbo", family_name="Baggins") + with pytest.raises(Exception, match="Existence query is not specific enough"): + person.save(mock_client, space="myspace", recursive=False) + assert len(mock_client.instances) == 3 + + +def test_save_retrieves_all_matches_if_more_than_query_size(mock_client, clear_caches, monkeypatch): + """ + The existence query asks for only a few results. If more instances match, they are all retrieved, + so that an instance in the target space is not missed. + """ + for i in range(EXISTENCE_QUERY_SIZE + 1): + _seed_person(mock_client, f"collab-{i}", uuid=f"00000000-0000-0000-0000-00000000000{i}") + target_id = _seed_person(mock_client, "myspace", uuid="23456789-0abc-def0-1234-567890abcdef") + + sizes = [] + original_query = mock_client.query + + def paginated_query(*args, size=100, **kwargs): + sizes.append(size) + response = original_query(*args, size=size, **kwargs) + total = response.total + response.data = response.data[:size] + response.total = total + return response + + monkeypatch.setattr(mock_client, "query", paginated_query) + person = omcore.Person(given_name="Bilbo", family_name="Baggins") + person.save(mock_client, space="myspace", recursive=False, ignore_duplicates=True) + assert sizes == [EXISTENCE_QUERY_SIZE, EXISTENCE_QUERY_SIZE + 2] + assert person.id == target_id + + +def test_exists_connection_lost(mock_client, clear_caches, monkeypatch): + def disconnected_query(*args, **kwargs): + raise ConnectionError( + ("Connection aborted.", RemoteDisconnected("Remote end closed connection without response")) + ) + + monkeypatch.setattr(mock_client, "query", disconnected_query) + person = omcore.Person(given_name="Bilbo", family_name="Baggins") + with pytest.warns(UserWarning, match="Timeout when checking for existence"): + assert not person.exists(mock_client) + + +def test_exists_other_connection_error(mock_client, clear_caches, monkeypatch): + def refused_query(*args, **kwargs): + raise ConnectionError("Failed to establish a new connection: [Errno 61] Connection refused") + + monkeypatch.setattr(mock_client, "query", refused_query) + person = omcore.Person(given_name="Bilbo", family_name="Baggins") + with pytest.raises(ConnectionError, match="Connection refused"): + person.exists(mock_client) + with pytest.raises(ConnectionError, match="Connection refused"): + person.save(mock_client, space="myspace", recursive=False) + + +def test_save_duplicates_in_other_spaces(mock_client, clear_caches): + first_id = _seed_person(mock_client, "common") + _seed_person(mock_client, "collab-foo", uuid="23456789-0abc-def0-1234-567890abcdef") + + person = omcore.Person(given_name="Bilbo", family_name="Baggins") + with pytest.raises(Exception, match="Existence query is not specific enough"): + person.save(mock_client, space="myspace", recursive=False) + assert len(mock_client.instances) == 2 + + person = omcore.Person(given_name="Bilbo", family_name="Baggins") + with pytest.warns(UserWarning, match="already exists in space 'common'"): + person.save(mock_client, space="myspace", recursive=False, ignore_duplicates=True) + assert person.id == first_id + assert len(mock_client.instances) == 2 + + # def test_save_existing_with_id_mock(mock_client): # existing_model = mock_client.instances[] diff --git a/test/utils.py b/test/utils.py index 537d9ae5..f77fc447 100644 --- a/test/utils.py +++ b/test/utils.py @@ -212,7 +212,11 @@ def _match_instances(self, query): if node_type not in as_list(instance.get("@type", [])): continue if all(self._value_matches(instance.get(path, None), op, value) for path, op, value in filters): - matches.append(deepcopy(instance)) + match = deepcopy(instance) + if "https://core.kg.ebrains.eu/vocab/meta/space" in match: + # the query API returns the space under the name given in the query definition + match["https://schema.hbp.eu/myQuery/space"] = match["https://core.kg.ebrains.eu/vocab/meta/space"] + matches.append(match) return matches @staticmethod