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