kapps_triplestore_interface.graph_db#

class kapps_triplestore_interface.graph_db.GraphDB(credentials: GraphDBCredentials, timeout: int = 60, use_gdb_token: bool = True, named_graph: str | IRI | None = None, logger: Logger | None = None)[source]#

Bases: object

High-level client for GraphDB repositories.

Provides convenience methods for querying and updating data, manages authentication, repository selection, a default named graph, and a Kafka connector manager.

classmethod from_env(logger: Logger | None = None) → GraphDB[source]#

Construct a client using environment-based credentials.

Returns:: A configured GraphDB instance using GraphDBCredentials.from_env().
Return type:: GraphDB

get_list_of_named_graphs() → List[str | IRI]#

Get the list of named graphs in the current repository.

Returns:: List of named graph IRIs. Can be an empty list.
Return type:: List[IRI]
Raises:: GraphDbException – If the underlying request to GraphDB fails.

fetch_statements(graph_iri: str | IRI | None = None) → Graph | None#

Fetch statements from a named or the default graph.

Queries the RDF4J Graph Store endpoint and parses the response into an rdflib.Graph. When graph_iri is None, the default graph is fetched.

Parameters:: graph_iri (Optional[GraphNameLike]) – The named graph IRI to fetch; when None, the default graph is fetched.
Returns:: The parsed graph on success; None when the request fails.
Return type:: Optional[Graph]
Raises:: requests.exceptions.RequestException – If the underlying HTTP request fails.

import_statements(content: str, overwrite: bool | None = False, graph_iri: str | IRI | None = None, content_type: str | None = 'application/x-turtle') → bool#

Import RDF statements into a named or the default graph.

Sends content to the RDF4J Graph Store endpoint using POST (append) or PUT (overwrite). When graph_iri is None, the default graph is targeted.

Parameters:

content (str) – RDF content to import.
overwrite (Optional[bool]) – Use PUT to overwrite existing content. Defaults to False.
graph_iri (Optional[GraphNameLike]) – Target named graph IRI; default graph when None.
content_type (Optional[str]) – MIME type of content (e.g., ‘application/x-turtle’). Defaults to ‘application/x-turtle’.

Returns:

True on success (HTTP 204), False otherwise.

Return type:

bool

Raises:

requests.exceptions.RequestException – If the underlying HTTP request fails.

clear_graph(graph_iri: str | IRI | None = None) → bool#

Clear a named graph or the default graph.

Deletes the specified named graph from the triplestore; when graph_iri is None, clears the default graph.

Parameters:: graph_iri (Optional[GraphNameLike]) – IRI of the named graph to clear; default graph when None.
Returns:: True on success (HTTP 204), False otherwise.
Return type:: bool
Raises:: requests.exceptions.RequestException – If the underlying HTTP request fails.

Check whether a specific triple exists in the graph database.

Parameters:

triple (TripleLike) – The triple (subject, predicate, object) to check.
named_graph (Optional[GraphNameLike]) – Override the client’s default named graph.

Returns:

True if the triple exists, False otherwise.

Return type:

bool

Add a triple to the graph database.

Parameters:

triple (TripleLike) – The triple (subject, predicate, object) to insert.
named_graph (Optional[GraphNameLike]) – Override the client’s default named graph.

Returns:

True if the triple was inserted, False otherwise.

Return type:

bool

Delete a single triple.

A SPARQL DELETE operation in GraphDB can be successful even if the triple does not exist. When check_exist=True, the function verifies the triple is present before attempting deletion.

Parameters:

triple (TripleLike) – The triple (subject, predicate, object) to delete.
check_exist (Optional[bool]) – Whether to verify existence prior to deletion. Defaults to True.
named_graph (Optional[GraphNameLike]) – Override the client’s default named graph.

Returns:

True if deletion succeeded (or triple absent when check_exist=False), False otherwise.

Return type:

bool

Update a triple by replacing any of its parts.

Performs a SPARQL DELETE … INSERT … WHERE that replaces the old triple with a new triple built from provided parts.

Parameters:

old_triple (TripleLike) – Existing triple to update.
new_triple (Optional[PartialTripleLike]) – Replacement values (subject/predicate/object). Use this or new_sub/new_pred/new_obj.
new_sub (Optional[SubjectLike]) – Replacement subject.
new_pred (Optional[PredicateLike]) – Replacement predicate.
new_obj (Optional[ObjectLike]) – Replacement object.
check_exist (Optional[bool]) – If True, verify that old_triple exists before updating. Defaults to True.
named_graph (Optional[GraphNameLike]) – Override the client’s default named graph.

Returns:

True if the update succeeded, False otherwise.

Return type:

bool

Raises:

InvalidInputError – If neither or both of new_triple and any of new_sub/new_pred/new_obj are provided, or if input triples are incomplete/invalid.

Retrieve triples matching any combination of subject, predicate, or object.

Parameters:

triple (Optional[PartialTripleLike]) – Combined (subject, predicate, object) filter tuple. Use this or individual sub/pred/obj.
sub (Optional[SubjectLike]) – Subject filter (IRI/shorthand/string).
pred (Optional[PredicateLike]) – Predicate filter (IRI/shorthand/string).
obj (Optional[ObjectLike]) – Object filter (IRI/shorthand/Literal/string).
include_explicit (Optional[bool]) – Include explicit triples. Defaults to True.
include_implicit (Optional[bool]) – Include inferred triples. Defaults to True.
named_graph (Optional[GraphNameLike]) – Override the client’s default named graph.

Returns:

Matching triples as (subject, predicate, object), where the object is converted to an appropriate Python type when applicable.

Return type:

List[Tuple[Subject, Predicate, Any]]

Raises:

InvalidInputError – If neither or both of triple and any of sub/pred/obj are provided.

Check if any of the given triples exist.

Parameters:

triples (TriplesLike) – Triples to check.
named_graph (Optional[GraphNameLike]) – Override the client’s default named graph.

Returns:

True if at least one exists, False otherwise.

Return type:

bool

Check if all of the given triples exist.

Parameters:

triples (TriplesLike) – Triples to check.
named_graph (Optional[GraphNameLike]) – Override the client’s default named graph.

Returns:

True if all exist, False otherwise.

Return type:

bool

Add multiple triples to the graph database.

Parameters:

triples_to_add (TriplesLike) – Triples to add.
check_exist (Optional[bool]) – If True, abort when any triple already exists. Defaults to True.
named_graph (Optional[GraphNameLike]) – Override the client’s default named graph.

Returns:

True if all triples were added, False otherwise.

Return type:

bool

Delete multiple triples from the graph database.

Parameters:

triples_to_delete (TriplesLike) – Triples to delete.
check_exist (Optional[bool]) – If True, abort when any triple does not exist. Defaults to True.
named_graph (Optional[GraphNameLike]) – Override the client’s default named graph.

Returns:

True if all triples were deleted, False otherwise.

Return type:

bool

Update multiple RDF triples in the triplestore.

The removal of old_triples and the insertion of new_triples are applied in a single atomic DELETE … INSERT … WHERE SPARQL transaction. The two lists need not be the same length: this supports pure additions, pure removals, and general replacements. Atomicity matters for constraint (e.g. SHACL) correctness — replacing a cardinality-constrained property (such as a possession handover under a “possessed by exactly one resource” shape) must never expose the intermediate state in which the property is momentarily absent.

Parameters:

old_triples (TriplesLike) – Triples to remove.
new_triples (TriplesLike) – Triples to insert. Need not match the length of old_triples.
check_exist (Optional[bool]) – If True, abort when any old triple does not exist. Defaults to True.
named_graph (Optional[GraphNameLike]) – Override the client’s default named graph.

Returns:

True if the update was successful, False otherwise.

Return type:

bool

Check if an IRI exists as subject, predicate, or object.

Parameters:

iri (IRILike) – The IRI to check for existence.
as_sub (Optional[bool]) – If True, check existence as subject. Defaults to False.
as_pred (Optional[bool]) – If True, check existence as predicate. Defaults to False.
as_obj (Optional[bool]) – If True, check existence as object. Defaults to False.
include_explicit (Optional[bool]) – Include explicit triples (FROM onto:explicit). Defaults to True.
include_implicit (Optional[bool]) – Include inferred triples (FROM onto:implicit). Defaults to True.
named_graph (Optional[GraphNameLike]) – Override the client’s default named graph.

Returns:

True if the IRI exists based on the specified criteria, False otherwise.

Return type:

bool

Raises:

InvalidInputError – If none of as_sub, as_pred, or as_obj is True.

new_iri(base: str | IRI, schema: Callable[[IRI], IRI] | None = None, test_schema: bool = True) → IRI#

Generate a new unique IRI within the graph database’s namespace.

Parameters:

base (IRILike) – The base IRI or namespace for the new IRI.
schema (Optional[Callable[[IRI], IRI]]) – A callable that generates the new IRI. Takes the base IRI as an argument. Defaults to a {onto}#{fragment}-{UUID4}. If fragment of base is empty, uses {onto}#instance-{UUID4}.

Returns:

A new unique IRI.

Return type:

IRI

new_blank_id(schema: Callable[[], str] | None=<function <lambda>>) → str#

Generate a new unique blank node identifier.

Parameters:: schema (Optional[Callable[[], str]]) – A callable that generates the blank node ID. Defaults to a genid-<uuid4> format.
Returns:: A new unique blank node identifier.
Return type:: str

is_subclass(subclass_iri: str | IRI, class_iri: str | IRI, named_graph: str | IRI | None = None) → bool#

Check whether one class is a subclass of another (rdfs:subClassOf).

Asks for subclass_iri rdfs:subClassOf class_iri

Parameters:

subclass_iri (IRILike) – The IRI of the potential subclass.
class_iri (IRILike) – The IRI of the potential superclass.
named_graph (Optional[GraphNameLike]) – Override the client’s default named graph.

Returns:

True if subclass_iri is a subclass of class_iri, False otherwise.

Return type:

bool

owl_is_named_individual(iri: str | IRI, named_graph: str | IRI | None = None) → bool#

Check if the given IRI corresponds to an OWL named individual.

Asks for iri rdf:type owl:NamedIndividual.

Parameters:

iri (IRILike) – The IRI to check.
named_graph (Optional[GraphNameLike]) – Override the client’s default named graph.

Returns:

True if the IRI is a named individual, False otherwise.

Return type:

bool

Get the OWL classes associated with a given individual.

Builds a SPARQL query that returns the classes for an instance IRI and optionally filters out results by prefix or returns local names only.

Parameters:

instance_iri (IRILike) – IRI of the individual to inspect.
ignored_prefixes (Optional[List[Namespace]]) – Prefixes/namespaces to ignore when collecting classes. Defaults to [“owl”, “rdfs”].
local_name (Optional[bool]) – If True, return only the local names of the classes. Defaults to False.
include_explicit (Optional[bool]) – Include explicit triples. Defaults to True.
include_implicit (Optional[bool]) – Include inferred triples. Defaults to False.
named_graph (Optional[GraphNameLike]) – Override the client’s default named graph.

Returns:

Class IRIs, or local names if local_name=True.

Return type:

List[Union[IRI, str]]

property repository: str#

The currently selected repository identifier.

Returns:: The active repository id.
Return type:: str

property named_graph: IRI | None#

The currently selected default named graph.

Returns:: The default named graph as an IRI, or None if unset.
Return type:: Optional[IRI]

property named_graph_str: str | None#

The selected default named graph as a string.

Returns:: The IRI string of the default named graph, or None.
Return type:: Optional[str]

get_list_of_repositories(only_ids: bool | None = False) → List[str] | List[dict] | None[source]#

List repositories available on the GraphDB instance.

Parameters:: only_ids (Optional[bool]) – When True, return only the repository ids. When False, return the full repository descriptor objects. Defaults to False.
Returns:: A list of ids if only_ids=True, a list of repository descriptors otherwise, or None when the request fails.
Return type:: Union[List[str], List[dict], None]

close() → None[source]#: Release the HTTP connection pools held for every thread that used this client.

Execute a SPARQL query or update against the repository.

Parameters:

query (Union[SPARQLQuery, str]) – The SPARQL query/update string to execute.
update (Optional[bool]) – If True, perform an update; otherwise perform a read query. Defaults to False.
convert_bindings (Optional[bool]) – Whether to convert query result bindings to Python types. Defaults to True.

Returns:

If update is False, the parsed JSON result dict. If update is True, True on success. Returns None or False only when failures are handled upstream; otherwise an exception is raised.

Return type:

Optional[Union[Dict, bool]]

Raises:

TypeError – If the query parameter is neither a SPARQLQuery nor a string.
InvalidQueryError – If the SPARQL query is malformed.
GraphDbException – If the HTTP request succeeds but the GraphDB API returns an error status.