rdf4j

Classes:

• RDF4JStore –

  An RDF4J store.

RDF4JStore

RDF4JStore(base_url: str, repository_id: str, configuration: str | None = None, auth: tuple[str, str] | None = None, timeout: float = 30.0, create: bool = False, **kwargs)

Bases: Store

An RDF4J store.

Methods:

• __len__ –
• add –
• addN –
• add_graph –
• bind –
• close –
• contexts –
• namespace –
• namespaces –
• open –
• prefix –
• remove –
• remove_graph –
• triples –

Attributes:

• client –
• context_aware –
• formula_aware –
• graph_aware –
• repo –
• transaction_aware –

Source code in rdflib/plugins/stores/rdf4j.py

def __init__(
    self,
    base_url: str,
    repository_id: str,
    configuration: str | None = None,
    auth: tuple[str, str] | None = None,
    timeout: float = 30.0,
    create: bool = False,
    **kwargs,
):
    if configuration is None:
        configuration = _default_repo_config(repository_id)
    self._client = RDF4JClient(base_url, auth, timeout, **kwargs)
    self._repository_id = repository_id
    self._repo = None
    self.open(configuration, create)
    super().__init__()

client property

client

context_aware class-attribute instance-attribute

context_aware = True

formula_aware class-attribute instance-attribute

formula_aware = False

graph_aware class-attribute instance-attribute

graph_aware = True

repo property

repo

transaction_aware class-attribute instance-attribute

transaction_aware = False

__len__

__len__(context: _ContextType | None = None) -> int

Source code in rdflib/plugins/stores/rdf4j.py

def __len__(self, context: _ContextType | None = None) -> int:
    return self.repo.size(context if context is None else context.identifier)

add

add(triple: _TripleType, context: _ContextType | None = None, quoted: bool = False) -> None

Source code in rdflib/plugins/stores/rdf4j.py

def add(
    self,
    triple: _TripleType,
    context: _ContextType | None = None,
    quoted: bool = False,
) -> None:
    s, p, o = triple
    graph_name = (
        ""
        if context is None or context.identifier == DATASET_DEFAULT_GRAPH_ID
        else context.identifier.n3()
    )
    statement = f"{s.n3()} {p.n3()} {o.n3()} {graph_name} ."
    self.repo.upload(statement)

addN

addN(quads: Iterable[_QuadType]) -> None

Source code in rdflib/plugins/stores/rdf4j.py

def addN(self, quads: Iterable[_QuadType]) -> None:  # noqa: N802
    statements = ""
    for s, p, o, c in quads:
        graph_name = (
            ""
            if c is None or c.identifier == DATASET_DEFAULT_GRAPH_ID
            else c.identifier.n3()
        )
        statement = f"{s.n3()} {p.n3()} {o.n3()} {graph_name} .\n"
        statements += statement
    self.repo.upload(statements)

add_graph

add_graph(graph: Graph) -> None

Source code in rdflib/plugins/stores/rdf4j.py

def add_graph(self, graph: Graph) -> None:
    if graph.identifier != DATASET_DEFAULT_GRAPH_ID:
        # Note: this is a no-op since RDF4J doesn't support empty named graphs.
        self.repo.update(f"CREATE SILENT GRAPH {graph.identifier.n3()}")

bind

bind(prefix: str, namespace: URIRef, override: bool = True) -> None

Source code in rdflib/plugins/stores/rdf4j.py

def bind(self, prefix: str, namespace: URIRef, override: bool = True) -> None:
    # Note: RDF4J namespaces always override.
    self.repo.namespaces.set(prefix, namespace)

close

close(commit_pending_transaction: bool = False) -> None

Source code in rdflib/plugins/stores/rdf4j.py

def close(self, commit_pending_transaction: bool = False) -> None:
    self.client.close()

contexts

contexts(triple: Optional[_TripleType] = None) -> Generator[_ContextType, None, None]

Source code in rdflib/plugins/stores/rdf4j.py

def contexts(
    self, triple: Optional[_TripleType] = None
) -> Generator[_ContextType, None, None]:
    if triple is None:
        for graph_name in self.repo.graph_names():
            yield Graph(self, identifier=graph_name)
    else:
        s, p, o = triple
        params = (
            _node_to_sparql(s if s else Variable("s")),
            _node_to_sparql(p if p else Variable("p")),
            _node_to_sparql(o if o else Variable("o")),
        )
        query = (
            "SELECT DISTINCT ?graph WHERE { GRAPH ?graph { %s %s %s } }" % params
        )
        result = self.repo.query(query)
        for row in result:
            yield Graph(self, identifier=row["graph"])

namespace

namespace(prefix: str) -> Optional[URIRef]

Source code in rdflib/plugins/stores/rdf4j.py

def namespace(self, prefix: str) -> Optional[URIRef]:
    result = self.repo.namespaces.get(prefix)
    return URIRef(result) if result is not None else None

namespaces

namespaces() -> Iterator[Tuple[str, URIRef]]

Source code in rdflib/plugins/stores/rdf4j.py

def namespaces(self) -> Iterator[Tuple[str, URIRef]]:
    for result in self.repo.namespaces.list():
        yield result.prefix, URIRef(result.namespace)

open

open(configuration: str | tuple[str, str] | None, create: bool = False) -> int | None

Source code in rdflib/plugins/stores/rdf4j.py

def open(
    self, configuration: str | tuple[str, str] | None, create: bool = False
) -> int | None:
    try:
        # Try connecting to the repository.
        self.repo.health()
    except RepositoryNotFoundError:
        if create:
            self.client.repositories.create(self._repository_id, configuration)
            self.repo.health()
        else:
            raise Exception(f"Repository {self._repository_id} not found.")

    return VALID_STORE

prefix

prefix(namespace: URIRef) -> Optional[str]

Source code in rdflib/plugins/stores/rdf4j.py

def prefix(self, namespace: URIRef) -> Optional[str]:
    namespace_prefixes = dict(
        [(x.namespace, x.prefix) for x in self.repo.namespaces.list()]
    )
    return namespace_prefixes.get(str(namespace))

remove

remove(triple: _TriplePatternType, context: Optional[_ContextType] = None) -> None

Source code in rdflib/plugins/stores/rdf4j.py

def remove(
    self,
    triple: _TriplePatternType,
    context: Optional[_ContextType] = None,
) -> None:
    s, p, o = triple
    g = context.identifier if context is not None else None
    self.repo.delete(s, p, o, g)

remove_graph

remove_graph(graph: Graph) -> None

Source code in rdflib/plugins/stores/rdf4j.py

def remove_graph(self, graph: Graph) -> None:
    self.repo.graphs.clear(graph.identifier)

triples

triples(triple_pattern: _TriplePatternType, context: Optional[_ContextType] = None) -> Iterator[Tuple[_TripleType, Iterator[Optional[_ContextType]]]]

Source code in rdflib/plugins/stores/rdf4j.py

def triples(
    self,
    triple_pattern: _TriplePatternType,
    context: Optional[_ContextType] = None,
) -> Iterator[Tuple[_TripleType, Iterator[Optional[_ContextType]]]]:
    s, p, o = triple_pattern
    graph_name = context.identifier if context is not None else None
    result_graph = self.repo.get(s, p, o, graph_name)
    if isinstance(result_graph, Dataset):
        for s, p, o, g in result_graph:
            yield (s, p, o), iter([Graph(self, identifier=g)])
    else:
        # It's a Graph object.
        for triple in result_graph:
            # Returning None for _ContextType as it's not used by the caller.
            yield triple, iter([None])