graphdb

Modules:

• client –

  GraphDB client module.

• exceptions –

  GraphDB exceptions.

• models –
• util –

Classes:

• GraphDBClient –

  GraphDB Client

__all__ module-attribute

__all__ = ['GraphDBClient']

GraphDBClient

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

Bases: RDF4JClient

GraphDB Client

This client and its inner management objects perform HTTP requests via httpx and may raise httpx-specific exceptions. Errors documented by GraphDB in its OpenAPI 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 GraphDB 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:

• login –

  Authenticate with GraphDB and obtain a GDB token.

Attributes:

• cluster (ClusterGroupManagement) –
• graphdb_repositories (RepositoryManagement) –
• monitoring (MonitoringManagement) –
• recovery (RecoveryManagement) –
• repositories (RepositoryManager) –

  Server-level repository management operations (GraphDB-specific).

• security (SecurityManagement) –
• users (UserManagement) –

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

def __init__(
    self,
    base_url: str,
    auth: tuple[str, str] | str | None = None,
    timeout: float | httpx.Timeout = 30.0,
    **kwargs: t.Any,
):
    super().__init__(base_url, auth, timeout, **kwargs)
    self._cluster_group = ClusterGroupManagement(self.http_client)
    self._monitoring = MonitoringManagement(self.http_client)
    self._recovery = RecoveryManagement(self.http_client)
    self._graphdb_repository_manager = RepositoryManager(self.http_client)
    self._graphdb_repositories = RepositoryManagement(self.http_client)
    self._security = SecurityManagement(self.http_client)
    self._users = UserManagement(self.http_client)

cluster property

cluster: ClusterGroupManagement

graphdb_repositories property

graphdb_repositories: RepositoryManagement

monitoring property

monitoring: MonitoringManagement

recovery property

recovery: RecoveryManagement

repositories property

repositories: RepositoryManager

Server-level repository management operations (GraphDB-specific).

security property

security: SecurityManagement

users property

users: UserManagement

login

login(username: str, password: str) -> AuthenticatedUser

Authenticate with GraphDB and obtain a GDB token.

Parameters:

• username

  (str) –

  The username to authenticate with.

• password

  (str) –

  The password to authenticate with.

Returns:

• AuthenticatedUser –

  An AuthenticatedUser instance containing user details and the GDB token.

Raises:

• UnauthorisedError –

  If the credentials are invalid.

• BadRequestError –

  If the request body is invalid.

• ResponseFormatError –

  If the response cannot be parsed.

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

def login(self, username: str, password: str) -> AuthenticatedUser:
    """Authenticate with GraphDB and obtain a GDB token.

    Parameters:
        username: The username to authenticate with.
        password: The password to authenticate with.

    Returns:
        An AuthenticatedUser instance containing user details and the GDB token.

    Raises:
        UnauthorisedError: If the credentials are invalid.
        BadRequestError: If the request body is invalid.
        ResponseFormatError: If the response cannot be parsed.
    """
    try:
        response = self.http_client.post(
            "/rest/login",
            json={"username": username, "password": password},
        )
        response.raise_for_status()

        auth_header = response.headers.get("Authorization", "")
        if not auth_header.startswith("GDB "):
            raise ResponseFormatError(
                "Authorization header missing or invalid format"
            )

        try:
            data = response.json()
        except (ValueError, TypeError) as err:
            raise ResponseFormatError(
                f"Failed to parse login response: {err}"
            ) from err

        try:
            return AuthenticatedUser.from_response(data, auth_header)
        except (ValueError, TypeError) as err:
            raise ResponseFormatError(
                f"Failed to parse authenticated user: {err}"
            ) from err

    except httpx.HTTPStatusError as err:
        status = err.response.status_code
        if status == 400:
            raise BadRequestError(f"Invalid request: {err.response.text}") from err
        elif status == 401:
            raise UnauthorisedError(
                f"Invalid credentials: {err.response.text}"
            ) from err
        raise