rdf4j

Modules:

• client –

  RDF4J client module.

• exceptions –

  RDF4J exceptions.

• util –

  RDF4J utility functions.

Classes:

• RDF4JClient –

  RDF4J client.

Attributes:

• has_httpx –

__all__ module-attribute

__all__ = ['RDF4JClient', 'has_httpx']

has_httpx module-attribute

has_httpx = find_spec('httpx') is not None

RDF4JClient

RDF4JClient(base_url: str, auth: tuple[str, str] | str | None = None, timeout: float | Timeout = 30.0, **kwargs: Any)

RDF4J client.

This client and its inner management objects perform HTTP requests via httpx and may raise httpx-specific exceptions. Errors documented by RDF4J in its protocol specification are mapped to specific exceptions in this library where applicable. Error mappings are documented on each management method. The underlying httpx client is reused across requests, and connection pooling is handled automatically by httpx.

Parameters:

• base_url

  (str) –

  The base URL of the RDF4J server.

• auth

  (tuple[str, str] | str | None, default: None ) –

  Authentication credentials. Can be a tuple (username, password) for basic auth, or a string for token-based auth (e.g., “GDB “) which is added as the Authorization header.

• timeout

  (float | Timeout, default: 30.0 ) –

  Request timeout in seconds or an httpx.Timeout for fine-grained control (default: 30.0).

• kwargs

  (Any, default: {} ) –

  Additional keyword arguments to pass to the httpx.Client.

Methods:

• __enter__ –
• __exit__ –
• close –

  Close the underlying httpx.Client.

Attributes:

• http_client –
• protocol (float) –

  The RDF4J REST API protocol version.

• repositories (RepositoryManager) –

  Server-level repository management operations.

Source code in rdflib/contrib/rdf4j/client.py

def __init__(
    self,
    base_url: str,
    auth: tuple[str, str] | str | None = None,
    timeout: float | httpx.Timeout = 30.0,
    **kwargs: Any,
):
    if not base_url.endswith("/"):
        base_url += "/"

    httpx_auth: tuple[str, str] | None = None
    if isinstance(auth, tuple):
        httpx_auth = auth
    elif isinstance(auth, str):
        headers = kwargs.get("headers", {})
        headers["Authorization"] = auth
        kwargs["headers"] = headers

    self._http_client = httpx.Client(
        base_url=base_url, auth=httpx_auth, timeout=timeout, **kwargs
    )
    self._repository_manager = RepositoryManager(self.http_client)
    try:
        protocol_version = self.protocol
    except httpx.RequestError as err:
        self.close()
        raise RDF4JUnsupportedProtocolError(
            f"Failed to check protocol version: {err}"
        ) from err
    if protocol_version < 12:
        self.close()
        raise RDF4JUnsupportedProtocolError(
            f"RDF4J server protocol version {protocol_version} is not supported. Minimum required version is 12."
        )

http_client property

http_client

protocol property

protocol: float

The RDF4J REST API protocol version.

Returns:

• float –

  The protocol version number.

repositories property

repositories: RepositoryManager

Server-level repository management operations.

__enter__

__enter__()

Source code in rdflib/contrib/rdf4j/client.py

def __enter__(self):
    return self

__exit__

__exit__(exc_type, exc_val, exc_tb)

Source code in rdflib/contrib/rdf4j/client.py

def __exit__(self, exc_type, exc_val, exc_tb):
    self.close()

close

close()

Close the underlying httpx.Client.

Source code in rdflib/contrib/rdf4j/client.py

def close(self):
    """Close the underlying httpx.Client."""
    self.http_client.close()