"""
This module defines the KGClient class, which provides a thin interface
on top of the kg-core-python package, for communicating with the
EBRAINS KG core API.
"""
# Copyright 2018-2026 CNRS and fairgraph authors and/or their employers
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations
import os
import logging
import re
from typing import Any, Dict, Iterable, List, Optional, Union, TYPE_CHECKING
from uuid import uuid4, UUID
try:
from kg_core.kg import kg
from kg_core.request import Stage, Pagination, ExtendedResponseConfiguration, ReleaseTreeScope
from kg_core.response import ResultPage, JsonLdDocument, SpaceInformation, Error
have_kg_core = True
except ImportError:
have_kg_core = False
from openminds.registry import lookup_type
from .errors import AuthenticationError, AuthorizationError, ResourceExistsError
from .utility import handle_scope_keyword
from .base import OPENMINDS_VERSION
if TYPE_CHECKING:
from .kgobject import KGObject
try:
import clb_nb_utils.oauth as clb_oauth # type: ignore
except ImportError:
clb_oauth = None
logger = logging.getLogger("fairgraph")
if have_kg_core:
STAGE_MAP = {
"released": Stage.RELEASED,
"latest": Stage.IN_PROGRESS,
"in progress": Stage.IN_PROGRESS,
}
default_response_configuration = ExtendedResponseConfiguration(return_embedded=True)
BARE_UUID = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}")
def expand_bare_uuids(data: Any, namespace: str) -> Any:
"""
Replace every "@id" whose value is a bare UUID with the full URI of the instance, in place.
Marmotgraph v3 gives links to other KG instances as full URIs
(e.g. "https://kg.ebrains.eu/api/instances/<uuid>"), but Marmotgraph v4 gives them as bare UUIDs,
while still giving the "@id" of the instance itself as a full URI. Expanding the bare UUIDs keeps
ids consistent, so that links can be resolved and compared with the ids of the instances they
point to. Responses containing only full URIs are unchanged.
Returns the (modified) data.
"""
if isinstance(data, dict):
for key, value in data.items():
if key == "@id":
# checking the length first quickly rules out full URIs and other IRIs
if isinstance(value, str) and len(value) == 36 and BARE_UUID.fullmatch(value):
data[key] = f"{namespace}{value}"
else:
expand_bare_uuids(value, namespace)
elif isinstance(data, list):
for item in data:
expand_bare_uuids(item, namespace)
return data
AVAILABLE_PERMISSIONS = [
"CREATE",
"READ",
"DELETE",
"RELEASE",
"INVITE_FOR_REVIEW",
"INVITE_FOR_SUGGESTION",
"SUGGEST",
"UNRELEASE",
"READ_RELEASED",
"RELEASE_STATUS",
"WRITE",
]
[docs]
class KGClient(object):
"""
A client for accessing the EBRAINS Knowledge Graph (KG) API.
It can be used to retrieve, add, update, and delete KG nodes.
Attributes:
cache (dict): A dictionary used for caching JSON-LD documents.
accepted_terms_of_use (bool): A boolean indicating whether the user has accepted the terms of use.
Args:
token (str, optional): An EBRAINS authentication token for accessing the KG API.
host (str, optional): The hostname of the KG API. Use "core.kg-ppd.ebrains.eu" for testing
and "core.kg.ebrains.eu" to work with the production KG.
client_id (str, optional): For use together with client_secret in place of the token if you have a service account.
client_secret (str, optional): The client secret to use for authentication. Required if client_id is provided.
allow_interactive (bool, default True): if true, allow authentication via web browser
openminds_version (str, default "v4"): the openMINDS schema version that responses should be
deserialized into. Must be one of "v4" or "v5". v4 is the default so existing code is
unaffected; pass "v5" when connecting to a KG instance that has been migrated to v5.
Raises:
ImportError: If the kg_core package is not installed.
AuthenticationError: If neither a token nor client ID/secret are provided.
ValueError: If openminds_version is not "v4" or "v5".
"""
def __init__(
self,
token: Optional[str] = None,
host: str = "core.kg-ppd.ebrains.eu",
client_id: Optional[str] = None,
client_secret: Optional[str] = None,
allow_interactive: bool = True,
openminds_version: str = OPENMINDS_VERSION,
):
if openminds_version not in ("v4", "v5"):
raise ValueError(
f"openminds_version must be 'v4' or 'v5', got {openminds_version!r}"
)
self.openminds_version = openminds_version
if not have_kg_core:
raise ImportError("Please install the ebrains-kg-core package")
if client_id and client_secret:
self._kg_client_builder = kg(host).with_credentials(client_id, client_secret)
self._auth_method = "credentials"
elif token:
self._kg_client_builder = kg(host).with_token(token)
self._auth_method = "token"
elif clb_oauth:
self._kg_client_builder = kg(host).with_token(clb_oauth.get_token()) # running in EBRAINS Jupyter Lab
self._auth_method = "clb_oauth"
else:
try:
self._kg_client_builder = kg(host).with_token(os.environ["KG_AUTH_TOKEN"])
self._auth_method = "env"
except KeyError:
if allow_interactive:
iam_config_url = "https://iam.ebrains.eu/auth/realms/hbp/.well-known/openid-configuration"
self._kg_client_builder = kg(host).with_device_flow(
client_id="kg-core-python", open_id_configuration_url=iam_config_url
)
self._auth_method = "device_flow"
else:
raise AuthenticationError("Need to provide either token or client id/secret.")
self._kg_client = self._kg_client_builder.build()
self.__kg_admin_client = None
self.host = host
self._user_info = None
self.cache: Dict[str, JsonLdDocument] = {}
self._query_cache: Dict[str, str] = {}
self.accepted_terms_of_use = False
if allow_interactive:
self.user_info()
[docs]
def refresh(self):
"""Rebuild the client with a fresh authentication token.
For credentials and device_flow auth, this obtains a new token automatically.
For clb_oauth (Jupyter Lab), this requests a fresh token from the platform.
For static tokens (passed directly or via env var), this prompts the user
to paste a new token interactively.
"""
if self._auth_method in ("token", "env"):
new_token = input("Authentication token has expired. Please paste a new token: ").strip()
if not new_token:
raise AuthenticationError("No token provided.")
self._kg_client_builder = kg(self.host).with_token(new_token)
elif self._auth_method == "clb_oauth":
self._kg_client_builder = kg(self.host).with_token(clb_oauth.get_token())
self._kg_client = self._kg_client_builder.build()
self.__kg_admin_client = None
self._user_info = None
@property
def _kg_admin_client(self):
if self.__kg_admin_client is None:
self.__kg_admin_client = self._kg_client_builder.build_admin()
return self.__kg_admin_client
@property
def token(self) -> Optional[str]:
return self._kg_client.instances._kg_config.token_handler._fetch_token()
def _check_response(
self,
response: ResultPage[JsonLdDocument],
ignore_not_found: bool = False,
error_context: str = "",
expected_instance_id: Optional[str] = None,
id_key: str = "@id",
) -> ResultPage[JsonLdDocument]:
if expected_instance_id and not response.error:
if response.total > 1:
# if an instance_id is specified, we expect there to be only one result
# if there are more, it seems to mean that the instance_id does not exist
response.error = Error(
code=404,
message=(
f"Received multiple results when specifying instance_id {expected_instance_id}"
"This indicates the instance does not exist."
),
)
response.data = []
response.size = response.total = 0
else:
if response.size == 1:
if str(expected_instance_id) not in response.data[0][id_key]:
raise Exception("mismatched instance_id")
if response.error:
# todo: handle "ignore_not_found"
if response.error.code == 403:
raise AuthorizationError(f"{response} {error_context}")
elif response.error.code == 401:
raise AuthenticationError(f"{response} {error_context}")
elif response.error.code == 404 and ignore_not_found:
return response
elif response.error.code == 409:
raise ResourceExistsError(f"{response} {error_context}")
else:
raise Exception(f"Error: {response.error} {error_context}")
else:
expand_bare_uuids(response.data, self._kg_client.instances._kg_config.id_namespace)
return response
[docs]
def query(
self,
query: Dict[str, Any],
filter: Optional[Dict[str, str]] = None,
instance_id: Optional[str] = None,
from_index: int = 0,
size: int = 100,
release_status: str = "released",
scope: Optional[str] = None,
id_key: str = "@id",
use_stored_query: bool = False,
restrict_to_spaces: Optional[List[str]] = None,
) -> ResultPage[JsonLdDocument]:
"""
Execute a Knowledge Graph (KG) query with the given filters and query definition.
Args:
query (Dict[str, Any]): A dictionary containing the query definition in JSON-LD.
filter (Dict[str, str]): A dictionary of filters to apply to the query. Each key represents the prop name to filter on,
and the value represents the value(s) to filter on.
instance_id (Optional[URI]): The URI of a specific KG instance to retrieve.
from_index (int): The index of the first result to return (0-based).
size (int): The maximum number of results to return.
release_status (str): The scope of the query. Valid values are "released", "in progress", or "any". Default is "released".
id_key (str): The key that identifies the ID of a JSON-LD document. Default is "@id".
use_stored_query (bool): Whether to use a stored query with the given query_id instead of a dynamic query. Default is False.
Returns:
A ResultPage object containing a list of JSON-LD instances that satisfy the query,
along with metadata about the query results such as total number of instances, and pagination information.
Raises:
ValueError: if a value in `filter` contains "+" or "%" (see below).
"""
release_status = handle_scope_keyword(scope, release_status)
if filter:
# `filter` values are sent to the KG as request parameters, which the KG decodes twice:
# once by Spring, and again in DataQueryBuilder.createAqlForFilter() in marmotgraph-core.
# As a result, a "+" is received as a space, so the query silently returns the wrong results,
# and a "%" either causes a "400 Bad Request" error or is decoded together with the following
# characters. Filter values given within the query definition itself are not affected.
# The second decoding is absent from the v4 branch of marmotgraph-core, which replaces DataQueryBuilder.
# If test_kg_misreads_plus_and_percent_in_query_parameters in test/test_client.py starts failing,
# the KG has been fixed and this check can be removed.
for name, value in filter.items():
values = value if isinstance(value, (list, tuple)) else [value]
if any(isinstance(item, str) and ("+" in item or "%" in item) for item in values):
raise ValueError(
f"Cannot filter on {name}={value!r} using a query parameter, since the KG does not handle "
"'+' or '%' in parameter values correctly. Include the filter value in the query definition "
"instead."
)
query_id = query.get("@id", None)
if use_stored_query:
def _query(release_status, from_index, size):
response = self._kg_client.queries.execute_query_by_id(
query_id=self.uuid_from_uri(query_id),
additional_request_params=filter or {},
stage=STAGE_MAP[release_status],
pagination=Pagination(start=from_index, size=size),
instance_id=instance_id,
restrict_to_spaces=restrict_to_spaces,
)
error_context = f"_query(release_status={release_status} query_id={query_id} filter={filter} instance_id={instance_id} size={size} from_index={from_index})"
return self._check_response(
response, error_context=error_context, expected_instance_id=instance_id, id_key=id_key, ignore_not_found=True
)
else:
def _query(release_status, from_index, size):
response = self._kg_client.queries.test_query(
query,
additional_request_params=filter or {},
stage=STAGE_MAP[release_status],
pagination=Pagination(start=from_index, size=size),
instance_id=instance_id,
restrict_to_spaces=restrict_to_spaces,
)
error_context = f"_query(release_status={release_status} query_id={query_id} filter={filter} instance_id={instance_id} size={size} from_index={from_index})"
return self._check_response(
response, error_context=error_context, expected_instance_id=instance_id, id_key=id_key, ignore_not_found=True
)
if release_status == "any":
# the following implementation is simple but very inefficient
# because we retrieve _all_ instances and then apply the limits
# from_index and size.
# todo: make this more efficient, but be sure to clearly
# explain the algorithm
instances = {}
# first we get the released instances
response = _query("released", 0, 100000)
for instance in response.data:
instances[instance[id_key]] = instance
# now we get the "in progress" instances, and overwrite
# any existing released instances which have the same id
response = _query("in progress", 0, 100000)
for instance in response.data:
instances[instance[id_key]] = instance
response.data = list(instances.values())[from_index : from_index + size]
response.size = len(response.data)
response.total = len(instances)
else:
response = _query(release_status, from_index, size)
return response
[docs]
def list(
self,
target_type: str,
space: Optional[str] = None,
from_index: int = 0,
size: int = 100,
release_status: str = "released",
scope: Optional[str] = None,
) -> ResultPage[JsonLdDocument]:
"""
List KG instances of a given type.
Args:
target_type: The URI if the instance type to list.
space: If specified, restricts the search to the given space.
from_index: The index of the first result to include in the response.
size: The maximum number of results to include in the response.
release_status: The scope of instances to include in the response. Valid values are
'released', 'in progress', 'any'. If 'any' is specified, all accessible instances
are included in the response, but this may be slow where there are large numbers of instances.
Returns:
A ResultPage object containing the list of JSON-LD instances,
along with metadata about the query results such as total number of instances, and pagination information.
"""
release_status = handle_scope_keyword(scope, release_status)
def _list(release_status, from_index, size):
response = self._kg_client.instances.list(
stage=STAGE_MAP[release_status],
target_type=target_type,
space=space,
response_configuration=default_response_configuration,
pagination=Pagination(start=from_index, size=size),
)
error_context = f"_list(release_status={release_status} space={space} target_type={target_type} size={size} from_index={from_index})"
return self._check_response(response, error_context=error_context)
if release_status == "any":
# see comments in query() about this implementation
instances = {}
# first we get the released instances
response = _list("released", 0, 100000)
for instance in response.data:
instances[instance["@id"]] = instance
# now we get the "in progress" instances, and overwrite
# any existing released instances which have the same id
response = _list("in progress", 0, 100000)
for instance in response.data:
instances[instance["@id"]] = instance
response.data = list(instances.values())[from_index : from_index + size]
response.size = len(response.data)
response.total = len(instances)
return response
else:
return _list(release_status, from_index, size)
[docs]
def instance_from_full_uri(
self,
uri: str,
use_cache: bool = True,
release_status: str = "released",
scope: Optional[str] = None,
require_full_data: bool = True,
) -> JsonLdDocument:
"""
Return a specific KG instance identified by its URI.
Args:
uri: The global identifier of the instance
use_cache: whether to use cached data if they exist. Defaults to True.
release_status: The release_status of instances to include in the response.
Valid values are 'released', 'in progress', 'any'.
require_full_data: Whether to only return instances for which the user has full read access.
"""
release_status = handle_scope_keyword(scope, release_status)
logger.debug("Retrieving instance from {}, api='core' use_cache={}".format(uri, use_cache))
data: JsonLdDocument
if use_cache and uri in self.cache:
logger.debug("Retrieving instance {} from cache".format(uri))
data = self.cache[uri]
else:
def _get_instance(release_status):
error_context = f"_get_instance(release_status={release_status} uri={uri})"
# Normal KG URIs start with https://kg.ebrains.eu/api/instances/ with a UUID
# but for openMINDS controlled terms we may have the openMINDS URI
# of the form https://openminds.om-i.org/instances/ageCategory/juvenile
# We use different query methods for these different cases.
kg_namespace = self._kg_client.instances._kg_config.id_namespace
if uri.startswith(kg_namespace):
response = self._kg_client.instances.get_by_id(
stage=STAGE_MAP[release_status],
instance_id=self.uuid_from_uri(uri),
extended_response_configuration=default_response_configuration,
)
response = self._check_response(response, error_context=error_context, ignore_not_found=True)
if response.error:
assert response.error.code == 404 # all other errors should have been trapped by the check
data = None
else:
data = response.data
elif uri.startswith("https://openminds.om-i.org/instances"):
payload = [uri]
response = self._kg_client.instances.get_by_identifiers(
stage=STAGE_MAP[release_status],
payload=payload,
extended_response_configuration=default_response_configuration,
)
# todo: handle errors
data = expand_bare_uuids(response.data[payload[0]].data, kg_namespace)
else:
raise Exception(f"This client cannot retrieve instances from {uri}")
# in some circumstances, the KG returns "minimal" metadata,
# e.g. with just the id and fullName properties
# this means the user does not have full access, so we count this as no data
if require_full_data and data and "http://schema.org/identifier" not in data:
data = None
return data
if release_status == "any":
data_ip = _get_instance("in progress")
data_rel = _get_instance("released")
data = data_rel or data_ip
if data_ip is not None:
data.update(data_ip)
else:
data = _get_instance(release_status)
if data:
self.cache[uri] = data
return data
[docs]
def create_new_instance(
self, data: JsonLdDocument, space: str, instance_id: Optional[str] = None
) -> JsonLdDocument:
"""
Create a new KG instance using the data provided.
Args:
data (dict): a JSON-LD document that should be added to the KG as a new instance.
space (str): the space in which the instance should be stored.
instance_id (UUID, optional): a UUID that should be used as the basis for the
instance's persistent identifier. If not specified, the KG will generate an ID.
"""
if "'@id': None" in str(data):
raise ValueError("payload contains undefined ids")
if instance_id:
UUID(instance_id)
if instance_id:
response = self._kg_client.instances.create_new_with_id(
space=space,
payload=data,
instance_id=instance_id,
extended_response_configuration=default_response_configuration,
)
else:
response = self._kg_client.instances.create_new(
space=space,
payload=data,
extended_response_configuration=default_response_configuration,
)
error_context = f"create_new_instance(data={data}, space={space}, instance_id={instance_id})"
return self._check_response(response, error_context=error_context).data
[docs]
def update_instance(self, instance_id: str, data: JsonLdDocument) -> JsonLdDocument:
"""
Update an existing KG instance using the data provided.
Args:
instance_id (UUID): the instance's persistent identifier.
data (dict): a JSON-LD document that modifies some or all of the data of the existing instance.
"""
UUID(instance_id)
response = self._kg_client.instances.contribute_to_partial_replacement(
instance_id=instance_id,
payload=data,
extended_response_configuration=default_response_configuration,
)
error_context = f"update_instance(data={data}, instance_id={instance_id})"
response_data = self._check_response(response, error_context=error_context).data
# The cached document for this URI is now stale. Drop it so the next
# fetch goes to the KG. We deliberately do not refresh the cache from
# the response, because depending on the server's `returnPayload`
# default the response may be a minimal document (e.g. just `@id`)
# rather than the full updated instance.
self.cache.pop(self.uri_from_uuid(instance_id), None)
return response_data
[docs]
def replace_instance(self, instance_id: str, data: JsonLdDocument) -> JsonLdDocument:
"""
Replace an existing KG instance using the data provided.
Args:
instance_id (UUID): the instance's persistent identifier.
data (dict): a JSON-LD document that will replace the existing instance.
"""
UUID(instance_id)
response = self._kg_client.instances.contribute_to_full_replacement(
instance_id=instance_id,
payload=data,
extended_response_configuration=default_response_configuration,
)
error_context = f"replace_instance(data={data}, instance_id={instance_id})"
response_data = self._check_response(response, error_context=error_context).data
# See update_instance for rationale: invalidate, don't refresh.
self.cache.pop(self.uri_from_uuid(instance_id), None)
return response_data
[docs]
def delete_instance(self, instance_id: str, ignore_not_found: bool = True, ignore_errors: bool = True):
"""
Delete a KG instance.
"""
if isinstance(instance_id, UUID):
instance_id = str(instance_id)
else:
UUID(instance_id)
response = self._kg_client.instances.delete(instance_id)
# response is None if no errors
if response: # error
if not ignore_errors:
raise Exception(response.message)
else:
self.cache.pop(self.uri_from_uuid(instance_id), None)
return response
[docs]
def uri_from_uuid(self, uuid: str) -> str:
"""Return an instance's URI given its UUID."""
namespace = self._kg_client.instances._kg_config.id_namespace
return f"{namespace}{uuid}"
[docs]
def uuid_from_uri(self, uri: str) -> UUID:
"""Return an instance's UUID given its URI."""
namespace = self._kg_client.instances._kg_config.id_namespace
assert uri.startswith(namespace)
return UUID(uri[len(namespace) :])
[docs]
def store_query(self, query_label: str, query_definition: Dict[str, Any], space: str):
"""
Store a query definition in the KG.
Args:
query_label (str): a label that can be used to identify and retrieve the query definition.
query_definition (dict): a JSON-LD document defining a KG query.
space (str): the space in which the query definition should be stored.
"""
existing_query = self.retrieve_query(query_label)
if existing_query:
query_id = self.uuid_from_uri(existing_query["@id"])
else:
query_id = uuid4()
try:
response = self._check_response(
self._kg_client.queries.save_query(
query_id=query_id, payload=query_definition, space=space or "myspace"
)
)
except AuthorizationError:
response = self._check_response(
self._kg_client.queries.save_query(query_id=query_id, payload=query_definition, space="myspace")
)
query_definition["@id"] = self.uri_from_uuid(query_id)
[docs]
def retrieve_query(self, query_label: str) -> Dict[str, Any]:
"""
Retrieve a stored query definition from the KG.
Args:
query_label (str): the label of the query definition to be retrieved.
"""
if query_label not in self._query_cache:
response = self._check_response(self._kg_client.queries.list_per_root_type(search=query_label))
if response.total == 0:
return None
elif response.total > 1:
# check for exact match to query_label
found_match = False
kgvq = "https://core.kg.ebrains.eu/vocab/query"
for result in response.data:
if result[f"{kgvq}/meta"][f"{kgvq}/name"] == query_label:
query_definition = result
found_match = True
break
if not found_match:
return None
else:
query_definition = response.data[0]
self._query_cache[query_label] = query_definition
return self._query_cache[query_label]
[docs]
def user_info(self) -> Dict[str, Any]:
"""
Returns information about the current user.
This information is that associated with the authorization token used.
"""
if self._user_info is None:
response = self._kg_client.users.my_info()
if response.data:
self._user_info = response.data
elif response.error.code == 401:
raise AuthenticationError()
elif response.error.code == 403:
raise AuthorizationError()
else:
raise Exception(response.error)
return self._user_info
def spaces(
self, permissions: Optional[Iterable[str] | bool] = None, names_only: bool = False
) -> Union[List[str], List[SpaceInformation]]:
f"""
Return a list of the Knowledge Graph spaces the user can access.
Args:
permissions (Optional[Iterable[str]]): Return only spaces for which the user has specific permissions.
The available permissions are as follows: {AVAILABLE_PERMISSIONS}
names_only (bool): Whether to return detailed information about each space (default) or only the space names.
Returns:
Union[List[str], List[SpaceInformation]]: either a list of SpaceInformation objects (default) or a list of names.
Raises:
ValueError: If an invalid permission string is included in permissions.
"""
if permissions and isinstance(permissions, Iterable):
for permission in permissions:
if permission.upper() not in AVAILABLE_PERMISSIONS:
raise ValueError(f"Invalid permission '{permission}'")
response = self._check_response(
self._kg_client.spaces.list(permissions=bool(permissions), pagination=Pagination(start=0, size=50))
)
accessible_spaces = list(response.items()) # makes additional requests if multiple pages of results
if permissions and isinstance(permissions, Iterable):
filtered_spaces = []
for space in accessible_spaces:
for permission in permissions:
if permission.upper() in space.permissions:
if names_only:
filtered_spaces.append(space.name)
else:
filtered_spaces.append(space)
return filtered_spaces
else:
if names_only:
return [space.name for space in accessible_spaces]
else:
return accessible_spaces
@property
def _private_space(self) -> str:
# temporary workaround
return f"private-{self.user_info().identifiers[0]}"
[docs]
def move_to_space(self, uri: str, destination_space: str):
"""
Move a KG instance from one space to another.
This is not recursive, i.e. child instances much be moved individually.
"""
# todo: add recursion
response = self._kg_client.instances.move(instance_id=self.uuid_from_uri(uri), space=destination_space)
if response.error:
raise Exception(response.error)
[docs]
def space_info(
self,
space_name: str,
release_status: str = "released",
scope: Optional[str] = None,
):
"""
Return information about the types and number of instances in a space.
The return format is a dictionary whose keys are classes and the values are the
number of instances of each class in the given spaces.
A space may contain types that have no fairgraph class, such as the KG's own
internal types. These are not an error: they appear in the result keyed by their
type IRI (a string) rather than by a class.
"""
release_status = handle_scope_keyword(scope, release_status)
result = self._kg_client.types.list(space=space_name, stage=STAGE_MAP[release_status])
if result.error:
raise Exception(result.error)
response = {}
for item in result.data:
type_iri = item.identifier
try:
cls = lookup_type(type_iri, self.openminds_version)
except ValueError:
cls = type_iri
response[cls] = item.occurrences
return response
[docs]
def clean_space(self, space_name):
"""Delete all instances from a given space."""
# todo: check for released instances, they must be unreleased
# before deletion.
space_info = self.space_info(space_name, release_status="in progress")
if sum(space_info.values()) > 0:
print(f"The space '{space_name}' contains the following instances:\n")
for cls, count in space_info.items():
if count > 0:
label = cls if isinstance(cls, str) else cls.__name__
print(label, count)
response = input("\nAre you sure you want to delete them? ")
if response not in ("y", "Y", "yes", "YES"):
return
error_messages = []
for cls, count in space_info.items():
if isinstance(cls, str):
target_type = cls
elif hasattr(cls, "list"): # exclude embedded metadata instances
target_type = cls.type_
else:
continue
if count > 0:
label = cls if isinstance(cls, str) else cls.__name__
print(f"Deleting {label} instances", end=" ")
response = self.list(target_type, release_status="in progress", space=space_name, size=count)
assert response.total <= count
for instance in response.data:
assert instance["https://core.kg.ebrains.eu/vocab/meta/space"] == space_name
error = self.delete_instance(self.uuid_from_uri(instance["@id"]), ignore_not_found=False)
if error:
print("x", end="")
error_messages.append(error)
else:
print(".", end="")
print()
if error_messages:
print(error_messages)
else:
print(f"The space '{space_name}' is already clean")
[docs]
def move_all_to_space(self, source_space: str, destination_space: str):
"""
Move all the KG instances in one space to another.
"""
assert source_space != destination_space
space_info = self.space_info(source_space, release_status="in progress")
if sum(space_info.values()) > 0:
print(f"The space '{source_space}' contains the following instances:\n")
for cls, count in space_info.items():
if count > 0:
label = cls if isinstance(cls, str) else cls.__name__
print(label, count)
response = input(f"\nAre you sure you want to move them to space '{destination_space}'? ")
if response not in ("y", "Y", "yes", "YES"):
return
for cls, count in space_info.items():
if count > 0:
if isinstance(cls, str):
label = cls
print(f"Moving {label} instances", end="")
response = self.list(cls, release_status="in progress", space=source_space, size=count)
for instance in response.data:
print(".", end="")
self.move_to_space(self.uuid_from_uri(instance["@id"]), destination_space)
print()
elif hasattr(cls, "list"): # exclude embedded metadata instances
print(f"Moving {cls.__name__} instances", end="")
instances = cls.list(self, release_status="in progress", space=source_space)
assert len(instances) <= count
for instance in instances:
assert instance.space == source_space
print(".", end="")
self.move_to_space(instance.id, destination_space)
print()
else:
print(f"The space '{source_space}' is empty, nothing to move.")
[docs]
def is_released(self, uri: str, with_children: bool = False) -> bool:
"""
Release status of a KG instance identified by its URI.
Args:
uri (URI): persistent identifier of the instance.
with_children (bool): whether to check if all the children of the instance
have also been released.
Returns:
True if the instance and (optionally) all its children have been released.
Otherwise False.
"""
if with_children:
release_tree_scope = ReleaseTreeScope.CHILDREN_ONLY
else:
release_tree_scope = ReleaseTreeScope.TOP_INSTANCE_ONLY
response = self._kg_client.instances.get_release_status(
instance_id=self.uuid_from_uri(uri), release_tree_scope=release_tree_scope
)
if response.data in ("RELEASED", "HAS_CHANGED"):
return True
elif response.data == "UNRELEASED":
return False
else:
raise AuthorizationError("You are not able to access the release status")
[docs]
def release(self, uri: str):
"""Release the instance with the given uri"""
response = self._kg_client.instances.release(self.uuid_from_uri(uri))
if response:
raise Exception(f"Can't release instance with id {uri}. Error message: {response}")
[docs]
def unrelease(self, uri: str):
"""Unrelease the instance with the given uri"""
response = self._kg_client.instances.unrelease(self.uuid_from_uri(uri))
if response:
raise Exception(f"Can't unrelease instance with id {uri}. Error message: {response}")
