"""GraphDB client module.""" from __future__ import annotations import json import os import pathlib import typing as t import httpx import rdflib.contrib.rdf4j from rdflib import Graph, Literal, URIRef from rdflib.contrib.graphdb.exceptions import ( BadRequestError, ConflictError, ForbiddenError, InternalServerError, NotFoundError, PreconditionFailedError, RepositoryNotFoundError, RepositoryNotHealthyError, ResponseFormatError, ServiceUnavailableError, UnauthorisedError, ) from rdflib.contrib.graphdb.models import ( AccessControlEntry, AuthenticatedUser, BackupOperationBean, ClearGraphAccessControlEntry, ClusterRequest, FreeAccessSettings, GraphDBRepository, ImportSettings, InfrastructureStatistics, NodeStatus, ParserSettings, PluginAccessControlEntry, RepositoryConfigBean, RepositoryConfigBeanCreate, RepositorySizeInfo, RepositoryStatistics, ServerImportBody, StatementAccessControlEntry, StructuresStatistics, SystemAccessControlEntry, User, UserCreate, UserUpdate, _parse_operation, _parse_plugin, _parse_policy, _parse_role, ) from rdflib.contrib.graphdb.util import ( extract_filename_from_content_disposition, infer_filename_from_fileobj, ) from rdflib.contrib.rdf4j import RDF4JClient FileContent = t.Union[ bytes, str, t.IO[bytes], t.Tuple[t.Optional[str], t.Union[bytes, str, t.IO[bytes]]], t.Tuple[t.Optional[str], t.Union[bytes, str, t.IO[bytes]], t.Optional[str]], ] FilesType = t.Union[ t.Mapping[str, FileContent], t.Iterable[t.Tuple[str, FileContent]], ] _ALLOWED_FGAC_SCOPES = {"statement", "clear_graph", "plugin", "system"} class AccessControlListManagement: """Manage fine-grained access control lists for GraphDB repositories.""" def __init__(self, identifier: str, http_client: httpx.Client): self._identifier = identifier self._http_client = http_client @property def http_client(self): return self._http_client @property def identifier(self): """Repository identifier.""" return self._identifier def list( self, scope: t.Literal["statement", "clear_graph", "plugin", "system"] | None = None, operation: t.Literal["read", "write", "*"] | None = None, subject: str | URIRef | None = None, predicate: str | URIRef | None = None, obj: str | URIRef | Literal | None = None, graph: t.Literal["*", "named", "default"] | URIRef | Graph | None = None, plugin: str | None = None, role: str | None = None, policy: t.Literal["allow", "deny", "abstain"] | None = None, ) -> t.List[ SystemAccessControlEntry | StatementAccessControlEntry | PluginAccessControlEntry | ClearGraphAccessControlEntry ]: """ List ACL rules for the repository. Parameters: scope: The scope of the FGAC rule (`statement`, `clear_graph`, `plugin`, `system`). operation: The operation of the FGAC rule (`read`, `write`, `*`). subject: The subject of the FGAC rule in Turtle format (for example, ``). predicate: The predicate of the FGAC rule in Turtle format (for example, ``). obj: The object of the FGAC rule in Turtle format (for example, `"Mary"@en`). graph: The graph of the FGAC rule in Turtle format (for example, ``). plugin: The plugin name for the FGAC rule with plugin scope (for example, `elasticsearch-connector`). role: The role associated with the FGAC rule. policy: The policy for the FGAC rule (`allow`, `deny`, `abstain`). Returns: A list of FGAC rules. Raises: ValueError: If the parameters are invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. InternalServerError: If the server returns an internal error. ResponseFormatError: If the response cannot be parsed. """ params: dict[str, str] = {} if scope is not None: if scope not in _ALLOWED_FGAC_SCOPES: raise ValueError(f"Invalid FGAC scope filter: {scope!r}") params["scope"] = scope if operation is not None: params["operation"] = _parse_operation(operation) if subject is not None: if isinstance(subject, URIRef): params["subject"] = subject.n3() elif isinstance(subject, str): params["subject"] = subject else: raise ValueError(f"Invalid FGAC subject filter: {subject!r}") if predicate is not None: if isinstance(predicate, URIRef): params["predicate"] = predicate.n3() elif isinstance(predicate, str): params["predicate"] = predicate else: raise ValueError(f"Invalid FGAC predicate filter: {predicate!r}") if obj is not None: if isinstance(obj, (URIRef, Literal)): params["object"] = obj.n3() elif isinstance(obj, str): params["object"] = obj else: raise ValueError(f"Invalid FGAC object filter: {obj!r}") if graph is not None: if isinstance(graph, URIRef): params["context"] = graph.n3() elif isinstance(graph, Graph): params["context"] = graph.identifier.n3() elif isinstance(graph, str): params["context"] = graph else: raise ValueError(f"Invalid FGAC graph filter: {graph!r}") if plugin is not None: params["plugin"] = _parse_plugin(plugin) if role is not None: params["role"] = _parse_role(role) if policy is not None: params["policy"] = _parse_policy(policy) try: response = self._http_client.get( f"/rest/repositories/{self.identifier}/acl", params=params ) response.raise_for_status() try: payload = response.json() except (ValueError, TypeError) as err: raise ResponseFormatError( f"Failed to parse GraphDB response: {err}" ) from err if not isinstance(payload, list): raise ResponseFormatError( "Failed to parse GraphDB response: expected a list of ACL entries." ) try: return [AccessControlEntry.from_dict(acl) for acl in payload] except (ValueError, TypeError) as err: raise ResponseFormatError( f"Failed to parse GraphDB response: {err}" ) from err except httpx.HTTPStatusError as err: status = err.response.status_code if status == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise def set(self, acl_rules: t.Sequence[AccessControlEntry]): """ Set ACL rules for the repository. !!! Note This method overwrites existing ACL rules in the repository. If you want to add new rules without overwriting existing ones, use the [`add`][rdflib.contrib.graphdb.client.AccessControlListManagement.add] method. Parameters: acl_rules: The list of ACL rules to set. Raises: BadRequestError: If the request is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. InternalServerError: If the server returns an internal error. ValueError: If the ACL rules are not provided as a sequence or are not AccessControlEntry instances. """ acl_rules_list = list(acl_rules) if any(not isinstance(rule, AccessControlEntry) for rule in acl_rules_list): raise ValueError("All ACL rules must be AccessControlEntry instances.") payload = [rule.as_dict() for rule in acl_rules_list] headers = {"Content-Type": "application/json"} try: response = self._http_client.put( f"/rest/repositories/{self.identifier}/acl", headers=headers, json=payload, ) response.raise_for_status() 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"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise def add( self, acl_rules: t.Sequence[AccessControlEntry], position: int | None = None ): """ Add ACL rules to the repository. You can also provide an optional URL request parameter position that specifies the position of the rules to be added. The position is zero-based (0 is the first position). If the position parameter is not provided, the rules are added at the end of the list. Parameters: acl_rules: The list of ACL rules to add. position: The zero-based position to add the rules at. Raises: BadRequestError: If the request is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. InternalServerError: If the server returns an internal error. ValueError: If the position is not an integer or is a negative integer. ValueError: If the ACL rules are not provided as a sequence or are not AccessControlEntry instances. """ acl_rules_list = list(acl_rules) if any(not isinstance(rule, AccessControlEntry) for rule in acl_rules_list): raise ValueError("All ACL rules must be AccessControlEntry instances.") payload = [rule.as_dict() for rule in acl_rules_list] headers = {"Content-Type": "application/json"} params = {} if position is not None: if not isinstance(position, int): raise ValueError("Position must be an integer.") if position < 0: raise ValueError("Position must be a positive integer.") params["position"] = str(position) try: response = self._http_client.post( f"/rest/repositories/{self.identifier}/acl", headers=headers, json=payload, params=params, ) response.raise_for_status() 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"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise def delete(self, acl_rules: t.Sequence[AccessControlEntry]): """ Delete ACL rules from the repository. The provided FGAC rules are removed from the list regardless of their position. Parameters: acl_rules: The list of ACL rules to delete. Raises: BadRequestError: If the request is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. InternalServerError: If the server returns an internal error. ValueError: If the ACL rules are not provided as a sequence or are not AccessControlEntry instances. """ acl_rules_list = list(acl_rules) if any(not isinstance(rule, AccessControlEntry) for rule in acl_rules_list): raise ValueError("All ACL rules must be AccessControlEntry instances.") payload = [rule.as_dict() for rule in acl_rules_list] headers = {"Content-Type": "application/json"} try: response = self._http_client.request( method="DELETE", url=f"/rest/repositories/{self.identifier}/acl", headers=headers, json=payload, ) response.raise_for_status() 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"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise class ClusterGroupManagement: """Manage and monitor GraphDB cluster configurations.""" def __init__(self, http_client: httpx.Client): self._http_client = http_client @property def http_client(self): return self._http_client def truncate_log(self) -> None: """Truncate the GraphDB cluster log. The truncate log operation is used to free up storage space on all cluster nodes by clearing the current transaction log and removing cached recovery snapshots. Raises: ResponseFormatError: If the response is not in the expected format. BadRequestError: If the request is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. """ try: response = self.http_client.post("/rest/cluster/truncate-log") response.raise_for_status() 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"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err raise def get_config(self) -> ClusterRequest: """Get the GraphDB cluster configuration. Returns: Cluster configuration. Raises: ResponseFormatError: If the response is not in the expected format. UnauthorisedError: If the request is unauthorised. NotFoundError: If the cluster configuration is not found. InternalServerError: If the server returns an internal error. """ try: headers = {"Accept": "application/json"} response = self.http_client.get("/rest/cluster/config", headers=headers) response.raise_for_status() try: return ClusterRequest.from_dict(response.json()) except (KeyError, TypeError, ValueError) as err: raise ResponseFormatError( f"Failed to parse cluster configuration: {err}" ) from err except httpx.HTTPStatusError as err: status = err.response.status_code if status == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif status == 404: raise NotFoundError( f"Cluster configuration not found: {err.response.text}" ) from err elif status == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise def create_config( self, nodes: list[str], election_min_timeout: int | None = None, election_range_timeout: int | None = None, heartbeat_interval: int | None = None, message_size_kb: int | None = None, verification_timeout: int | None = None, transaction_log_maximum_size_gb: int | None = None, batch_update_interval: int | None = None, ) -> None: """Create a GraphDB cluster. Parameters: nodes: List of node addresses. election_min_timeout: The minimum wait time in milliseconds for a heartbeat from a leader. election_range_timeout: The variable portion of each waiting period in milliseconds for a heartbeat. heartbeat_interval: The interval in milliseconds between each heartbeat that is sent to follower nodes by the leader. message_size_kb: The size of the data blocks, in kilobytes, transferred during data replication streaming through the RPC protocol. verification_timeout: The amount of time in milliseconds that a follower node would wait before attempting to verify the last committed entry when the first verification is unsuccessful. transaction_log_maximum_size_gb: Maximum size of the transaction log in GBs. The transaction log will be automatically truncated if it becomes bigger than this value. The minimum transaction log size is 1 GB. Setting it to a negative value will disable automatic transaction log truncation. batch_update_interval: The interval in milliseconds between requesting updates from the primary cluster. Used only when the cluster is in secondary mode. Raises: BadRequestError: If the request is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. ConflictError: If the cluster configuration already exists. PreconditionFailedError: If the cluster is not in a state to create a configuration. """ try: if not isinstance(nodes, list) or any( not isinstance(node, str) for node in nodes ): raise TypeError("nodes must be a list[str].") _int_fields: list[tuple[str, int | None]] = [ ("election_min_timeout", election_min_timeout), ("election_range_timeout", election_range_timeout), ("heartbeat_interval", heartbeat_interval), ("message_size_kb", message_size_kb), ("verification_timeout", verification_timeout), ("transaction_log_maximum_size_gb", transaction_log_maximum_size_gb), ("batch_update_interval", batch_update_interval), ] for field_name, value in _int_fields: if value is not None and type(value) is not int: raise TypeError(f"{field_name} must be an int if provided.") payload: dict[str, t.Any] = {"nodes": nodes} if election_min_timeout is not None: payload["electionMinTimeout"] = election_min_timeout if election_range_timeout is not None: payload["electionRangeTimeout"] = election_range_timeout if heartbeat_interval is not None: payload["heartbeatInterval"] = heartbeat_interval if message_size_kb is not None: payload["messageSizeKB"] = message_size_kb if verification_timeout is not None: payload["verificationTimeout"] = verification_timeout if transaction_log_maximum_size_gb is not None: payload["transactionLogMaximumSizeGB"] = transaction_log_maximum_size_gb if batch_update_interval is not None: payload["batchUpdateInterval"] = batch_update_interval headers = {"Content-Type": "application/json"} response = self.http_client.post( "/rest/cluster/config", headers=headers, json=payload ) response.raise_for_status() 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"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 409: raise ConflictError( f"Cluster configuration already exists: {err.response.text}" ) from err elif status == 412: raise PreconditionFailedError( f"Precondition failed: {err.response.text}" ) from err raise def delete_config(self, force: bool | None = None) -> dict[str, str]: """Delete the GraphDB cluster. By default, the cluster group cannot be deleted if one or more nodes are unreachable. Reachable here means that the nodes are not in status `NO_CONNECTION`, therefore there is an RPC connection to them. Parameter: force: When set to `True`, the cluster configuration will be deleted only on the reachable nodes and the response will always succeed. Returns: A dictionary where the node address is the key and the deletion status message is the value. Raises: ResponseFormatError: If the response is not in the expected format. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. PreconditionFailedError: If one or more nodes in the group are not reachable and force is not set to `True`. """ params: dict[str, str] | None = None if force is not None: params = {"force": str(force).lower()} try: headers = {"Accept": "application/json"} if params is None: response = self.http_client.delete( "/rest/cluster/config", headers=headers ) else: response = self.http_client.delete( "/rest/cluster/config", headers=headers, params=params ) response.raise_for_status() try: payload = response.json() except (ValueError, TypeError) as err: raise ResponseFormatError( f"Failed to parse cluster deletion response: {err}" ) from err if not isinstance(payload, dict): raise ResponseFormatError( "Failed to parse cluster deletion response: expected a JSON object." ) if not all( isinstance(key, str) and isinstance(value, str) for key, value in payload.items() ): raise ResponseFormatError( "Failed to parse cluster deletion response: expected string keys and values." ) return t.cast(dict[str, str], payload) except httpx.HTTPStatusError as err: status = err.response.status_code if status == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 412: raise PreconditionFailedError( f"Precondition failed: {err.response.text}" ) from err raise def update_config( self, election_min_timeout: int | None = None, election_range_timeout: int | None = None, heartbeat_interval: int | None = None, message_size_kb: int | None = None, verification_timeout: int | None = None, transaction_log_maximum_size_gb: int | None = None, batch_update_interval: int | None = None, ) -> ClusterRequest: """Update the GraphDB cluster configuration. If one of the cluster nodes is down or was not able to accept the new configuration, the operation will not be successful. A total consensus between the nodes is required. In case one or more of them cannot append the new configuration, it will be rejected by all of the nodes. Parameters: election_min_timeout: The minimum wait time in milliseconds for a heartbeat from a leader. election_range_timeout: The variable portion of each waiting period in milliseconds for a heartbeat. heartbeat_interval: The interval in milliseconds between each heartbeat that is sent to follower nodes by the leader. message_size_kb: The size of the data blocks, in kilobytes, transferred during data replication streaming through the RPC protocol. verification_timeout: The amount of time in milliseconds that a follower node would wait before attempting to verify the last committed entry when the first verification is unsuccessful. transaction_log_maximum_size_gb: Maximum size of the transaction log in GBs. The transaction log will be automatically truncated if it becomes bigger than this value. The minimum transaction log size is 1 GB. Setting it to a negative value will disable automatic transaction log truncation. batch_update_interval: The interval in milliseconds between requesting updates from the primary cluster. Used only when the cluster is in secondary mode. Returns: The updated cluster configuration. Raises: BadRequestError: If the request is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. PreconditionFailedError: If one or more nodes in the group are not reachable. """ _int_fields: list[tuple[str, int | None]] = [ ("election_min_timeout", election_min_timeout), ("election_range_timeout", election_range_timeout), ("heartbeat_interval", heartbeat_interval), ("message_size_kb", message_size_kb), ("verification_timeout", verification_timeout), ("transaction_log_maximum_size_gb", transaction_log_maximum_size_gb), ("batch_update_interval", batch_update_interval), ] for field_name, value in _int_fields: if value is not None and type(value) is not int: raise TypeError(f"{field_name} must be an int if provided.") payload: dict[str, t.Any] = {} if election_min_timeout is not None: payload["electionMinTimeout"] = election_min_timeout if election_range_timeout is not None: payload["electionRangeTimeout"] = election_range_timeout if heartbeat_interval is not None: payload["heartbeatInterval"] = heartbeat_interval if message_size_kb is not None: payload["messageSizeKB"] = message_size_kb if verification_timeout is not None: payload["verificationTimeout"] = verification_timeout if transaction_log_maximum_size_gb is not None: payload["transactionLogMaximumSizeGB"] = transaction_log_maximum_size_gb if batch_update_interval is not None: payload["batchUpdateInterval"] = batch_update_interval headers = {"Accept": "application/json"} try: response = self.http_client.patch( "/rest/cluster/config", headers=headers, json=payload ) response.raise_for_status() try: return ClusterRequest.from_dict(response.json()) except (KeyError, TypeError, ValueError) as err: raise ResponseFormatError( f"Failed to parse updated cluster configuration: {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"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 412: raise PreconditionFailedError( f"Precondition failed: {err.response.text}" ) from err raise def add_tag(self, tag: str) -> None: """Add primary cluster tag. Parameters: tag: The identifier tag to add to the primary cluster. Raises: TypeError: If tag is not a string. ValueError: If tag is an empty string. BadRequestError: If the request is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. PreconditionFailedError: If the cluster is not in a state to add a tag. """ if not isinstance(tag, str): raise TypeError("tag must be a string") if not tag: raise ValueError("tag must be a non-empty string") try: response = self.http_client.post( "/rest/cluster/config/tag", json={"tag": tag} ) response.raise_for_status() 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"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 412: raise PreconditionFailedError( f"Precondition failed: {err.response.text}" ) from err raise def delete_tag(self, tag: str) -> None: """Remove primary cluster tag. Parameters: tag: The identifier tag to remove from the primary cluster. Raises: TypeError: If tag is not a string. ValueError: If tag is an empty string. BadRequestError: If the request is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. PreconditionFailedError: If the cluster is not in a state to delete a tag. """ if not isinstance(tag, str): raise TypeError("tag must be a string") if not tag: raise ValueError("tag must be a non-empty string") headers = {"Content-Type": "application/json"} try: response = self.http_client.request( method="DELETE", url="/rest/cluster/config/tag", headers=headers, json={"tag": tag}, ) response.raise_for_status() 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"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 412: raise PreconditionFailedError( f"Precondition failed: {err.response.text}" ) from err raise def enable_secondary_mode(self, primary_node: str, tag: str) -> None: """Enable cluster secondary mode. You can switch any healthy primary cluster to secondary mode by providing the RPC address of the primary cluster node that it will replicate and then declaring the tag used to identify it. Once you have switched a cluster to secondary mode, all of its current repositories and data will be removed and replaced with the data and state of the primary cluster. !!! Note You can enable the secondary cluster mode from any of the nodes of a cluster, not just the leader node. You can also use any of the healthy nodes of the primary cluster to connect to that cluster. !!! Warning Enabling the secondary mode will delete all data on the secondary cluster and replicate the state of the primary cluster. Parameters: primary_node: The RPC address of one of the healthy nodes of the primary cluster. tag: The identifier tag of the primary cluster. Raises: TypeError: If primary_node or tag is not a string. ValueError: If primary_node or tag is an empty string. BadRequestError: If the request is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. PreconditionFailedError: If the cluster is not in a state to enable secondary mode. """ if not isinstance(primary_node, str): raise TypeError("primary_node must be a string") if not primary_node: raise ValueError("primary_node must be a non-empty string") if not isinstance(tag, str): raise TypeError("tag must be a string") if not tag: raise ValueError("tag must be a non-empty string") payload = {"primaryNode": primary_node, "tag": tag} try: response = self.http_client.post( "/rest/cluster/config/secondary-mode", json=payload ) response.raise_for_status() 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"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 412: raise PreconditionFailedError( f"Precondition failed: {err.response.text}" ) from err raise def disable_secondary_mode(self) -> None: """Disable cluster secondary mode. Once in secondary mode, you can disable secondary mode and then you will no longer be in that mode. Doing this will stop the pulling of updates from the primary cluster. Raises: BadRequestError: If the request is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. PreconditionFailedError: If the cluster is not in a state to disable secondary mode. """ try: response = self.http_client.delete("/rest/cluster/config/secondary-mode") response.raise_for_status() 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"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 412: raise PreconditionFailedError( f"Precondition failed: {err.response.text}" ) from err raise def add_nodes(self, nodes: list[str]) -> None: """Add nodes to the GraphDB cluster. Parameters: nodes: List of node addresses to add to the cluster. Raises: TypeError: If nodes is not a list of strings. BadRequestError: If the request is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. PreconditionFailedError: If one or more nodes in the group are not reachable. """ if not isinstance(nodes, list) or any( not isinstance(node, str) for node in nodes ): raise TypeError("nodes must be a list[str].") payload = {"nodes": nodes} headers = {"Content-Type": "application/json"} try: response = self.http_client.post( "/rest/cluster/config/node", headers=headers, json=payload ) response.raise_for_status() 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"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 412: raise PreconditionFailedError( f"Precondition failed: {err.response.text}" ) from err raise def remove_nodes(self, nodes: list[str]) -> None: """Remove nodes from the GraphDB cluster. Parameters: nodes: List of node addresses to remove from the cluster. Raises: TypeError: If nodes is not a list of strings. BadRequestError: If the request is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. PreconditionFailedError: If one or more nodes in the group are not reachable. """ if not isinstance(nodes, list) or any( not isinstance(node, str) for node in nodes ): raise TypeError("nodes must be a list[str].") payload = {"nodes": nodes} try: response = self.http_client.request( method="DELETE", url="/rest/cluster/config/node", json=payload, ) response.raise_for_status() 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"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 412: raise PreconditionFailedError( f"Precondition failed: {err.response.text}" ) from err raise def replace_nodes(self, add_nodes: list[str], remove_nodes: list[str]) -> None: """Replace nodes in the GraphDB cluster. Parameters: add_nodes: List of node addresses to add to the cluster. remove_nodes: List of node addresses to remove from the cluster. Raises: TypeError: If add_nodes or remove_nodes is not a list of strings. BadRequestError: If the request is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. PreconditionFailedError: If one or more nodes in the group are not reachable. """ if not isinstance(add_nodes, list) or any( not isinstance(node, str) for node in add_nodes ): raise TypeError("add_nodes must be a list[str].") if not isinstance(remove_nodes, list) or any( not isinstance(node, str) for node in remove_nodes ): raise TypeError("remove_nodes must be a list[str].") payload = {"removeNodes": remove_nodes, "addNodes": add_nodes} try: response = self.http_client.patch("/rest/cluster/config/node", json=payload) response.raise_for_status() 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"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 412: raise PreconditionFailedError( f"Precondition failed: {err.response.text}" ) from err raise def node_status(self) -> NodeStatus: """Get the status of this GraphDB cluster node. Returns: NodeStatus: The status of this GraphDB cluster node. Raises: ResponseFormatError: If the response cannot be parsed. BadRequestError: If the request is invalid. NotFoundError: If the node status is not found. InternalServerError: If the internal server error. """ try: headers = {"Accept": "application/json"} response = self.http_client.get( "/rest/cluster/node/status", headers=headers ) response.raise_for_status() try: return NodeStatus.from_dict(response.json()) except (KeyError, TypeError, ValueError) as err: raise ResponseFormatError( f"Failed to parse node status: {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 == 404: raise NotFoundError( f"Node status not found: {err.response.text}" ) from err elif status == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise def group_status(self) -> list[NodeStatus]: """Get the status of the GraphDB cluster. Returns: A list of node statuses for the cluster group. Raises: ResponseFormatError: If the response cannot be parsed. BadRequestError: If the request is invalid. NotFoundError: If the group status is not found. InternalServerError: If the internal server error. """ try: headers = {"Accept": "application/json"} response = self.http_client.get( "/rest/cluster/group/status", headers=headers ) response.raise_for_status() try: data = response.json() return [NodeStatus.from_dict(item) for item in data] except (KeyError, TypeError, ValueError) as err: raise ResponseFormatError( f"Failed to parse group status: {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 == 404: raise NotFoundError( f"Group status not found: {err.response.text}" ) from err elif status == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise class Repository(rdflib.contrib.rdf4j.client.Repository): """GraphDB Repository client. Overrides specific methods of the RDF4J Repository class and also provides GraphDB-only functionality (such as FGAC ACL management) made available through the GraphDB REST API. Parameters: identifier: The identifier of the repository. http_client: The httpx.Client instance. """ def __init__(self, identifier: str, http_client: httpx.Client): super().__init__(identifier, http_client) self._access_control_list_management = AccessControlListManagement( self.identifier, self.http_client ) @property def acl(self) -> AccessControlListManagement: return self._access_control_list_management def health(self, timeout: int = 5) -> bool: """Repository health check. Parameters: timeout: A timeout parameter in seconds. If provided, the endpoint attempts to retrieve the repository within this timeout. If not, the passive check is performed. Returns: bool: True if the repository is healthy, otherwise an error is raised. Raises: RepositoryNotFoundError: If the repository is not found. RepositoryNotHealthyError: If the repository is not healthy. """ try: params = {"passive": str(timeout)} response = self.http_client.get( f"/repositories/{self.identifier}/health", params=params ) response.raise_for_status() return True except httpx.HTTPStatusError as err: if err.response.status_code == 404: raise RepositoryNotFoundError( f"Repository {self._identifier} not found." ) raise RepositoryNotHealthyError( f"Repository {self._identifier} is not healthy. {err.response.status_code} - {err.response.text}" ) def get_server_import_files(self) -> list[ImportSettings]: """Get server files available for import. Returns: A list of files available for import. Raises: ResponseFormatError: If the response format is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. InternalServerError: If the server returns an internal error. """ try: headers = {"Accept": "application/json"} response = self.http_client.get( f"/rest/repositories/{self.identifier}/import/server", headers=headers, ) response.raise_for_status() try: result = [] for item in response.json(): # Convert nested parserSettings dict to ParserSettings instance if "parserSettings" in item and isinstance( item["parserSettings"], dict ): item["parserSettings"] = ParserSettings( **item["parserSettings"] ) result.append(ImportSettings(**item)) return result except (ValueError, TypeError) as err: raise ResponseFormatError(f"Failed to parse response: {err}") from err except httpx.HTTPStatusError as err: if err.response.status_code == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif err.response.status_code == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif err.response.status_code == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise def import_server_import_file(self, server_import_body: ServerImportBody) -> None: """Import a server file into the repository. Parameters: server_import_body: The server import body. Raises: BadRequestError: If the request is bad. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. NotFoundError: If the request is not found. """ try: headers = {"Content-Type": "application/json"} response = self.http_client.post( f"/rest/repositories/{self.identifier}/import/server", headers=headers, json=server_import_body.as_dict(), ) response.raise_for_status() except httpx.HTTPStatusError as err: status = err.response.status_code if status == 400: raise BadRequestError(f"Bad request: {err.response.text}") from err elif status == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 404: raise NotFoundError( f"Request is not found: {err.response.text}" ) from err raise def cancel_server_import_file(self, name: str) -> None: """Cancel server file import operation. Parameters: name: The name of the file import to interrupt. Raises: BadRequestError: If the request is bad. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. """ try: params = {"name": name} response = self.http_client.delete( f"/rest/repositories/{self.identifier}/import/server", params=params, ) response.raise_for_status() except httpx.HTTPStatusError as err: status = err.response.status_code if status == 400: raise BadRequestError(f"Bad request: {err.response.text}") from err elif status == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err raise class RepositoryManager(rdflib.contrib.rdf4j.client.RepositoryManager): """GraphDB Repository Manager. This manager client overrides specific RDF4J RepositoryManager class methods to provide GraphDB-specific implementations. """ def get(self, repository_id: str) -> Repository: """Get a repository by ID. !!! Note This performs a health check before returning the repository object. Parameters: repository_id: The identifier of the repository. Returns: Repository: The repository instance. Raises: RepositoryNotFoundError: If the repository is not found. RepositoryNotHealthyError: If the repository is not healthy. """ _repo = super().get(repository_id) return Repository(_repo.identifier, _repo.http_client) class MonitoringManagement: """ Monitor different GraphDB processes. """ def __init__(self, http_client: httpx.Client): self._http_client = http_client @property def http_client(self): return self._http_client @property def structures(self) -> StructuresStatistics: """Get structures statistics. Returns: The structures statistics. Raises: ResponseFormatError: If the response format is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. InternalServerError: If the server returns an internal error. ServiceUnavailableError: If the server is unavailable. """ try: headers = {"Accept": "application/json"} response = self.http_client.get("/rest/monitor/structures", headers=headers) response.raise_for_status() try: return StructuresStatistics(**response.json()) except (ValueError, TypeError) as err: raise ResponseFormatError(f"Failed to parse response: {err}") from err except httpx.HTTPStatusError as err: status = err.response.status_code if status == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err elif status == 503: raise ServiceUnavailableError( f"Service is unavailable: {err.response.text}" ) from err raise def get_repo_stats(self, repository_id: str): """Get repository statistics. Parameters: repository_id: The identifier of the repository. Returns: RepositoryStatistics: The repository statistics. Raises: ResponseFormatError: If the response cannot be parsed. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. InternalServerError: If the server returns an internal error. """ try: headers = {"Accept": "application/json"} response = self.http_client.get( f"/rest/monitor/repository/{repository_id}", headers=headers ) response.raise_for_status() try: return RepositoryStatistics.from_dict(response.json()) except (ValueError, TypeError, KeyError) as err: raise ResponseFormatError(f"Failed to parse response: {err}") from err except httpx.HTTPStatusError as err: status = err.response.status_code if status == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise @property def infrastructure(self) -> InfrastructureStatistics: """Get all infrastructure statistics. Returns: InfrastructureStatistics: The infrastructure statistics. Raises: ResponseFormatError: If the response cannot be parsed. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. InternalServerError: If the server returns an internal error. """ try: headers = {"Accept": "application/json"} response = self.http_client.get( "/rest/monitor/infrastructure", headers=headers ) response.raise_for_status() try: return InfrastructureStatistics.from_dict(response.json()) except (ValueError, TypeError, KeyError) as err: raise ResponseFormatError(f"Failed to parse response: {err}") from err except httpx.HTTPStatusError as err: status = err.response.status_code if status == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise def cluster(self) -> str: """Get cluster statistics. !!! Note This endpoint is only available in GraphDB EE. Returns: Prometheus-style metrics. Raises: UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. InternalServerError: If the server returns an internal error. ServiceUnavailableError: If the server is unavailable. """ try: headers = {"Accept": "text/plain"} response = self.http_client.get("/rest/monitor/cluster", headers=headers) response.raise_for_status() return response.text except httpx.HTTPStatusError as err: status = err.response.status_code if status == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err elif status == 503: raise ServiceUnavailableError( f"Service is unavailable: {err.response.text}" ) from err raise def backup(self) -> BackupOperationBean | None: """Track backup operations. Returns: On-going backup operations or `None` if no backup is in progress. Raises: ResponseFormatError: If the response cannot be parsed. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. InternalServerError: If the server returns an internal error. ServiceUnavailableError: If the server is unavailable. """ try: headers = {"Accept": "application/json"} response = self.http_client.get("/rest/monitor/backup", headers=headers) response.raise_for_status() try: data = response.json() if not data: return None return BackupOperationBean.from_dict(data) except (ValueError, TypeError, KeyError) as err: raise ResponseFormatError(f"Failed to parse response: {err}") from err except httpx.HTTPStatusError as err: status = err.response.status_code if status == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err elif status == 503: raise ServiceUnavailableError( f"Service is unavailable: {err.response.text}" ) from err raise class RecoveryManagement: """ GraphDB Recovery Management client. """ def __init__(self, http_client: httpx.Client): self._http_client = http_client @property def http_client(self): return self._http_client def restore( self, backup: bytes | bytearray | memoryview | t.IO[bytes] | str | os.PathLike[str], repositories: list[str] | None = None, restore_system_data: bool | None = None, remove_stale_repositories: bool | None = None, ) -> None: """Restore GraphDB instance from a backup archive. Parameters: backup: Backup archive content as bytes/bytes-like, a binary file-like object, or a filesystem path to a `.tar` produced by [`backup`][rdflib.contrib.graphdb.client.RecoveryManagement.backup]. repositories: List of repositories to restore. If `None` (default), the parameter is omitted and all repositories found in the backup are restored. If an empty list (`[]`) is provided, no repositories from the backup are restored. restore_system_data: Whether to restore system data (users, saved queries, visual graphs, etc.). If omitted, GraphDB defaults to `false`. If `true` and no system data exists in the backup, GraphDB returns an error. remove_stale_repositories: Whether to remove repositories on the target instance that are not restored from the backup. If omitted, GraphDB defaults to `false`. Raises: ValueError: If parameters are invalid, or if backup is not a supported type. BadRequestError: If the request is bad. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. """ if repositories is not None: if not isinstance(repositories, list): raise ValueError("repositories must be a list or None") if not all(isinstance(r, str) for r in repositories): raise ValueError("repositories must be a list of strings") if ( not isinstance(restore_system_data, bool) and restore_system_data is not None ): raise ValueError("restore_system_data must be a bool or None") if ( not isinstance(remove_stale_repositories, bool) and remove_stale_repositories is not None ): raise ValueError("remove_stale_repositories must be a bool or None") payload: dict[str, t.Any] = {} if repositories is not None: payload["repositories"] = repositories if restore_system_data is not None: payload["restoreSystemData"] = restore_system_data if remove_stale_repositories is not None: payload["removeStaleRepositories"] = remove_stale_repositories params_part: FileContent = (None, json.dumps(payload), "application/json") def _post(file_part: FileContent) -> None: try: response = self.http_client.post( "/rest/recovery/restore", files=[("params", params_part), ("file", file_part)], ) response.raise_for_status() except httpx.HTTPStatusError as err: status = err.response.status_code if status == 400: raise BadRequestError(f"Bad request: {err.response.text}") from err elif status == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err raise err if isinstance(backup, (bytes, bytearray, memoryview)): content = bytes(backup) _post(("graphdb-backup.tar", content, "application/x-tar")) return if hasattr(backup, "read") and not isinstance(backup, (str, os.PathLike)): file_obj = t.cast(t.IO[bytes], backup) filename = infer_filename_from_fileobj(file_obj, "graphdb-backup.tar") _post((filename, file_obj, "application/x-tar")) return if isinstance(backup, (str, os.PathLike)): backup_path = pathlib.Path(backup).expanduser() with open(backup_path, "rb") as f: filename = backup_path.name or infer_filename_from_fileobj( f, "graphdb-backup.tar" ) _post((filename, f, "application/x-tar")) return raise ValueError( "backup must be bytes/bytes-like, a binary file-like object, or a filesystem path" ) def cloud_restore( self, bucket_uri: str, repositories: list[str] | None = None, restore_system_data: bool | None = None, remove_stale_repositories: bool | None = None, authentication_file: ( bytes | bytearray | memoryview | t.IO[bytes] | str | os.PathLike[str] | None ) = None, ) -> None: """Restore GraphDB instance from a cloud backup archive. Parameters: bucket_uri: Cloud bucket URI including provider-specific parameters (required). repositories: List of repositories to restore. If `None` (default), the parameter is omitted and all repositories found in the backup are restored. If an empty list (`[]`) is provided, no repositories from the backup are restored. restore_system_data: Whether to restore system data (users, saved queries, visual graphs, etc.). If omitted, GraphDB defaults to `false`. If `true` and no system data exists in the backup, GraphDB returns an error. remove_stale_repositories: Whether to remove repositories on the target instance that are not restored from the backup. If omitted, GraphDB defaults to `false`. authentication_file: Optional credential file content as bytes/bytes-like, a binary file-like object, or a filesystem path. Raises: ValueError: If parameters are invalid, or authentication_file is not a supported type. BadRequestError: If the request is bad. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. """ if not isinstance(bucket_uri, str) or not bucket_uri: raise ValueError("bucket_uri must be a non-empty string") if repositories is not None: if not isinstance(repositories, list): raise ValueError("repositories must be a list or None") if not all(isinstance(r, str) for r in repositories): raise ValueError("repositories must be a list of strings") if ( not isinstance(restore_system_data, bool) and restore_system_data is not None ): raise ValueError("restore_system_data must be a bool or None") if ( not isinstance(remove_stale_repositories, bool) and remove_stale_repositories is not None ): raise ValueError("remove_stale_repositories must be a bool or None") payload: dict[str, t.Any] = {} if repositories is not None: payload["repositories"] = repositories if restore_system_data is not None: payload["restoreSystemData"] = restore_system_data if remove_stale_repositories is not None: payload["removeStaleRepositories"] = remove_stale_repositories payload["bucketUri"] = bucket_uri params_part: FileContent = (None, json.dumps(payload), "application/json") def _post(authentication_part: FileContent | None) -> None: files: list[tuple[str, FileContent]] = [("params", params_part)] if authentication_part is not None: files.append(("authenticationFile", authentication_part)) try: response = self.http_client.post( "/rest/recovery/cloud-restore", headers={"Accept": "application/json"}, files=files, ) response.raise_for_status() except httpx.HTTPStatusError as err: status = err.response.status_code if status == 400: raise BadRequestError(f"Bad request: {err.response.text}") from err elif status == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err raise err if authentication_file is None: _post(None) return if isinstance(authentication_file, (bytes, bytearray, memoryview)): content = bytes(authentication_file) _post(("authentication-file", content, "application/octet-stream")) return if hasattr(authentication_file, "read") and not isinstance( authentication_file, (str, os.PathLike) ): file_obj = t.cast(t.IO[bytes], authentication_file) filename = infer_filename_from_fileobj(file_obj, "authentication-file") _post((filename, file_obj, "application/octet-stream")) return if isinstance(authentication_file, (str, os.PathLike)): auth_path = pathlib.Path(authentication_file).expanduser() with open(auth_path, "rb") as f: filename = auth_path.name or infer_filename_from_fileobj( f, "authentication-file" ) _post((filename, f, "application/octet-stream")) return raise ValueError( "authentication_file must be bytes/bytes-like, a binary file-like object, or a filesystem path" ) @t.overload def backup( self, repositories: list[str] | None = None, backup_system_data: bool | None = None, *, dest: None = None, ) -> bytes: ... @t.overload def backup( self, repositories: list[str] | None = None, backup_system_data: bool | None = None, *, dest: str | os.PathLike[str], ) -> pathlib.Path: ... def backup( self, repositories: list[str] | None = None, backup_system_data: bool | None = None, *, dest: str | os.PathLike[str] | None = None, ) -> bytes | pathlib.Path: """Create a new backup of GraphDB instance. The backup is returned as a tar archive containing repository data and optionally system data (user accounts, saved queries, visual graphs, etc.). Parameters: repositories: List of repositories to be backed up, specified by their repository identifiers. If `None`, all repositories will be included in the backup. backup_system_data: Determines whether user account data such as user accounts, saved queries, or visual graphs, among others, should be included in the backup. dest: Destination path to save the backup archive. If `None`, the backup content is returned as bytes (suitable for small backups). If a directory path is provided, the filename from the response Content-Disposition header is used (e.g., backup-2026-01-16-10-30-00.tar). If a file path is provided, that exact path is used. The response is streamed directly to disk (recommended for large backups). Returns: bytes: If `dest` is `None`, returns the backup archive as bytes. Path: If `dest` is provided, returns the resolved `pathlib.Path` where the backup was saved. Raises: ValueError: If repositories is not a list or None, or if backup_system_data is not a bool or None. BadRequestError: If the request is bad. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. OSError: If the destination path cannot be written to. """ if repositories is not None: if not isinstance(repositories, list): raise ValueError("repositories must be a list or None") if not all(isinstance(r, str) for r in repositories): raise ValueError("repositories must be a list of strings") if not isinstance(backup_system_data, bool) and backup_system_data is not None: raise ValueError("backup_system_data must be a bool") payload: dict[str, t.Any] = {} if repositories is not None: payload["repositories"] = repositories if backup_system_data is not None: payload["backupSystemData"] = backup_system_data def _handle_http_error(err: httpx.HTTPStatusError) -> t.NoReturn: status = err.response.status_code if status == 400: raise BadRequestError(f"Bad request: {err.response.text}") from err elif status == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err raise err if dest is None: # Return bytes directly (suitable for small backups) try: response = self.http_client.post("/rest/recovery/backup", json=payload) response.raise_for_status() return response.content except httpx.HTTPStatusError as err: _handle_http_error(err) else: # Stream to disk (recommended for large backups) dest_path = pathlib.Path(dest).expanduser().resolve() try: with self.http_client.stream( "POST", "/rest/recovery/backup", json=payload ) as response: response.raise_for_status() # If dest is a directory, extract filename from Content-Disposition if dest_path.is_dir(): content_disposition = response.headers.get( "Content-Disposition", "" ) filename = extract_filename_from_content_disposition( content_disposition ) if filename: dest_path = dest_path / filename else: # Fallback to a default name if header parsing fails dest_path = dest_path / "graphdb-backup.tar" # Use a temporary file during download, then rename atomically tmp_path = dest_path.with_suffix(dest_path.suffix + ".partial") with open(tmp_path, "wb") as f: for chunk in response.iter_bytes(): f.write(chunk) tmp_path.replace(dest_path) return dest_path except httpx.HTTPStatusError as err: _handle_http_error(err) def cloud_backup( self, bucket_uri: str, repositories: list[str] | None = None, backup_system_data: bool | None = None, authentication_file: ( bytes | bytearray | memoryview | t.IO[bytes] | str | os.PathLike[str] | None ) = None, ) -> None: """Create a new backup of GraphDB instance uploaded to a cloud bucket. GraphDB uploads the resulting `.tar` archive directly to the configured cloud bucket. This uses the GraphDB endpoint `/rest/recovery/cloud-backup`. Parameters: bucket_uri: Cloud bucket URI including provider-specific parameters (required). repositories: Optional list of repositories to be backed up. backup_system_data: Optional flag to include system data (users, saved queries, etc.). authentication_file: Optional credential file content as bytes/bytes-like, a binary file-like object, or a filesystem path. Raises: ValueError: If parameters are invalid, or authentication_file is not a supported type. BadRequestError: If the request is bad. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. """ if not isinstance(bucket_uri, str) or not bucket_uri: raise ValueError("bucket_uri must be a non-empty string") if repositories is not None: if not isinstance(repositories, list): raise ValueError("repositories must be a list or None") if not all(isinstance(r, str) for r in repositories): raise ValueError("repositories must be a list of strings") if not isinstance(backup_system_data, bool) and backup_system_data is not None: raise ValueError("backup_system_data must be a bool") payload: dict[str, t.Any] = {} if repositories is not None: payload["repositories"] = repositories if backup_system_data is not None: payload["backupSystemData"] = backup_system_data payload["bucketUri"] = bucket_uri params_part: FileContent = (None, json.dumps(payload), "application/json") def _post(authentication_part: FileContent | None) -> None: files: list[tuple[str, FileContent]] = [("params", params_part)] if authentication_part is not None: files.append(("authenticationFile", authentication_part)) try: response = self.http_client.post( "/rest/recovery/cloud-backup", headers={"Accept": "application/json"}, files=files, ) response.raise_for_status() except httpx.HTTPStatusError as err: status = err.response.status_code if status == 400: raise BadRequestError(f"Bad request: {err.response.text}") from err elif status == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err raise err if authentication_file is None: _post(None) return if isinstance(authentication_file, (bytes, bytearray, memoryview)): content = bytes(authentication_file) _post(("authentication-file", content, "application/octet-stream")) return if hasattr(authentication_file, "read") and not isinstance( authentication_file, (str, os.PathLike) ): file_obj = t.cast(t.IO[bytes], authentication_file) filename = infer_filename_from_fileobj(file_obj, "authentication-file") _post((filename, file_obj, "application/octet-stream")) return if isinstance(authentication_file, (str, os.PathLike)): auth_path = pathlib.Path(authentication_file).expanduser() with open(auth_path, "rb") as f: filename = auth_path.name or infer_filename_from_fileobj( f, "authentication-file" ) _post((filename, f, "application/octet-stream")) return raise ValueError( "authentication_file must be bytes/bytes-like, a binary file-like object, or a filesystem path" ) class RepositoryManagement: """GraphDB Repository Management client. The functionality provided by this management client accepts an optional location parameter to operate on external GraphDB locations. """ def __init__(self, http_client: httpx.Client): self._http_client = http_client def list(self, location: str | None = None) -> list[GraphDBRepository]: """List all repositories. Parameters: location: The location of the repositories. Returns: list[GraphDBRepository]: List of GraphDB repositories. Raises: InternalServerError: If the server returns an internal error. ResponseFormatError: If the response cannot be parsed. """ params = {} if location is not None: params["location"] = location try: response = self.http_client.get("/rest/repositories", params=params) response.raise_for_status() try: return [GraphDBRepository.from_dict(repo) for repo in response.json()] except (ValueError, TypeError) as err: raise ResponseFormatError( f"Failed to parse GraphDB response: {err}" ) from err except httpx.HTTPStatusError as err: if err.response.status_code == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise def create( self, config: RepositoryConfigBeanCreate | str, location: str | None = None, files: FilesType | None = None, ): """Create a new repository. Parameters: config: Repository configuration. When a `RepositoryConfigBeanCreate` is provided, the request is sent as JSON. When a string is provided, it is treated as Turtle content and sent as multipart/form-data part `config` (required by GraphDB) with the content type `text/turtle`. location: Optional repository location (query param `location`). files: Optional extra multipart parts for GraphDB-specific files (e.g. `obdaFile`, `owlFile`, `propertiesFile`, `constraintFile`, `dbMetadataFile`, `lensesFile`). Keys must be the form part names; values may be file content or httpx-style file tuples. Ignored when `config` is a dataclass (JSON payload). Raises: BadRequestError: If the request is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. InternalServerError: If the server returns an internal error. """ params = {} if location is not None: params["location"] = location def _normalize_file_content(field: str, value: FileContent) -> FileContent: if isinstance(value, tuple): # fill in the missing filename with the field name if len(value) == 2: filename, content = value return (filename or field, content) filename, content, content_type = value return (filename or field, content, content_type) return (field, value) try: if isinstance(config, str): extra_parts: list[tuple[str, FileContent]] = [] if files is not None: if isinstance(files, t.Mapping): extra_parts = list(files.items()) else: extra_parts = list(files) if any(field == "config" for field, _ in extra_parts): raise ValueError( "Do not pass a 'config' multipart part via files; use the config argument instead." ) multipart_files: list[tuple[str, FileContent]] = [ ("config", ("config.ttl", config, "text/turtle")) ] for field, content in extra_parts: multipart_files.append( (field, _normalize_file_content(field, content)) ) response = self.http_client.post( "/rest/repositories", params=params, files=multipart_files, ) else: if files is not None: raise ValueError( "Additional files can only be provided when config is a Turtle string." ) response = self.http_client.post( "/rest/repositories", headers={"Content-Type": "application/json"}, json=config.as_dict(), params=params, ) response.raise_for_status() 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"Request is unauthorised: {err.response.text}" ) from err elif status == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif status == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise @property def http_client(self): return self._http_client @t.overload def get( self, repository_id: str, content_type: t.Literal["application/json"], location: str | None = None, ) -> RepositoryConfigBean: ... @t.overload def get( self, repository_id: str, content_type: None = None, location: str | None = None, ) -> RepositoryConfigBean: ... @t.overload def get( self, repository_id: str, content_type: t.Literal["text/turtle"], location: str | None = None, ) -> Graph: ... @t.overload def get( self, repository_id: str, content_type: str, location: str | None = None, ) -> RepositoryConfigBean | Graph: ... def get( self, repository_id: str, content_type: str | None = None, location: str | None = None, ) -> RepositoryConfigBean | Graph: """Get a repository's configuration. Parameters: repository_id: The identifier of the repository. content_type: The content type of the response. Can be `application/json` or `text/turtle`. Defaults to `application/json`. location: The location of the repository. Returns: RepositoryConfigBean: The repository configuration. Graph: The repository configuration in RDF. Raises: BadRequestError: If the content type is not supported. ResponseFormatError: If the response cannot be parsed. RepositoryNotFoundError: If the repository is not found. InternalServerError: If the server returns an internal error. """ if content_type is None: content_type = "application/json" if content_type not in ("application/json", "text/turtle"): raise ValueError(f"Unsupported content type: {content_type}.") headers = {"Accept": content_type} params = {} if location is not None: params["location"] = location try: response = self.http_client.get( f"/rest/repositories/{repository_id}", headers=headers, params=params ) response.raise_for_status() if content_type == "application/json": try: return RepositoryConfigBean(**response.json()) except (ValueError, TypeError) as err: raise ResponseFormatError( "Failed to parse GraphDB response." ) from err elif content_type == "text/turtle": try: return Graph().parse(data=response.text, format="turtle") except Exception as err: raise ResponseFormatError(f"Error parsing RDF: {err}") from err else: raise ValueError(f"Unhandled content type: {content_type}.") except httpx.HTTPStatusError as err: if err.response.status_code == 404: raise RepositoryNotFoundError( f"Repository {repository_id} not found." ) from err elif err.response.status_code == 500: raise InternalServerError("Internal server error.") from err raise def edit(self, repository_id: str, config: RepositoryConfigBeanCreate) -> None: """Edit a repository's configuration. Parameters: repository_id: The identifier of the repository. config: The repository configuration. Raises: BadRequestError: If the request is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. InternalServerError: If the server returns an internal error. """ headers = { "Content-Type": "application/json", } try: response = self.http_client.put( f"/rest/repositories/{repository_id}", headers=headers, json=config.as_dict(), ) response.raise_for_status() except httpx.HTTPStatusError as err: if err.response.status_code == 400: raise BadRequestError(f"Invalid request: {err.response.text}") from err elif err.response.status_code == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif err.response.status_code == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif err.response.status_code == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise def delete(self, repository_id: str, location: str | None = None) -> None: """Delete a repository. Parameters: repository_id: The identifier of the repository. location: The location of the repository. Raises: BadRequestError: If the request is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. InternalServerError: If the server returns an internal error. """ params = {} if location is not None: params["location"] = location try: response = self.http_client.delete( f"/rest/repositories/{repository_id}", params=params ) response.raise_for_status() except httpx.HTTPStatusError as err: if err.response.status_code == 400: raise BadRequestError(f"Invalid request: {err.response.text}") from err elif err.response.status_code == 401: raise UnauthorisedError( f"Request is unauthorised: {err.response.text}" ) from err elif err.response.status_code == 403: raise ForbiddenError( f"Request is forbidden: {err.response.text}" ) from err elif err.response.status_code == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise def restart( self, repository_id: str, sync: bool | None = None, location: str | None = None ) -> str: """Restart a repository. Parameters: repository_id: The identifier of the repository. sync: Whether to sync the repository. location: The location of the repository. Returns: str: The response text. Raises: UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. RepositoryNotFoundError: If the repository is not found. """ params = {} if sync is not None: params["sync"] = str(sync).lower() if location is not None: params["location"] = location try: response = self.http_client.post( f"/rest/repositories/{repository_id}/restart", params=params ) response.raise_for_status() return response.text except httpx.HTTPStatusError as err: if err.response.status_code == 401: raise UnauthorisedError("Request is unauthorised.") from err elif err.response.status_code == 403: raise ForbiddenError("Request is forbidden.") from err elif err.response.status_code == 404: raise RepositoryNotFoundError( f"Repository {repository_id} not found." ) from err raise def size( self, repository_id: str, location: str | None = None ) -> RepositorySizeInfo: """Get repository size. Parameters: repository_id: The identifier of the repository. location: The location of the repository. Raises: ResponseFormatError: If the response cannot be parsed. InternalServerError: If the server returns an internal error. """ params = {} if location: params["location"] = location try: response = self.http_client.get( f"/rest/repositories/{repository_id}/size", params=params ) response.raise_for_status() try: return RepositorySizeInfo(**response.json()) except (ValueError, TypeError) as err: raise ResponseFormatError("Failed to parse GraphDB response.") from err except httpx.HTTPStatusError as err: if err.response.status_code == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise @t.overload def validate( self, repository_id: str, *, content_type: str, content: str, location: str | None = None, shapes_repository_id: None = None, ) -> str: ... @t.overload def validate( self, repository_id: str, *, content: t.IO[bytes], content_type: str | None = None, location: str | None = None, shapes_repository_id: None = None, ) -> str: ... @t.overload def validate( self, repository_id: str, *, shapes_repository_id: str, location: str | None = None, ) -> str: ... def validate( self, repository_id: str, *, content_type: str | None = None, content: str | t.IO[bytes] | None = None, location: str | None = None, shapes_repository_id: str | None = None, ) -> str: """Validate repository data using SHACL shapes. Parameters: repository_id: The identifier of the repository. content_type: Content type of the request body (required for text payloads). content: SHACL shapes payload; string for text-based validation or file-like (binary) object for multipart validation. location: Optional repository location. shapes_repository_id: ID of repository containing SHACL shapes; when provided, no content is sent and shapes are fetched from that repository. Returns: str: Validation report as RDF Turtle. Raises: UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. InternalServerError: If the server returns an internal error. """ params = {} if location is not None: params["location"] = location if shapes_repository_id is not None: if content is not None: raise ValueError( "content must not be provided when shapes_repository_id is set." ) if content_type is not None: raise ValueError( "content_type must not be provided when shapes_repository_id is set." ) headers = {"Accept": "text/turtle"} try: response = self.http_client.post( url=f"/rest/repositories/{repository_id}/validate/repository/{shapes_repository_id}", params=params, headers=headers, ) response.raise_for_status() return response.text except httpx.HTTPStatusError as err: status = err.response.status_code if status == 401: raise UnauthorisedError("Request is unauthorised.") from err elif status == 403: raise ForbiddenError("Request is forbidden.") from err elif status == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise if content is None: raise ValueError("content must be provided.") is_file_like = hasattr(content, "read") and not isinstance(content, str) if is_file_like: headers = {"Accept": "text/turtle"} file_part = ( "shapes.ttl", t.cast(t.IO[bytes], content), content_type or "text/turtle", ) files: FilesType = {"file": file_part} try: response = self.http_client.post( url=f"/rest/repositories/{repository_id}/validate/file", params=params, headers=headers, files=files, ) response.raise_for_status() return response.text except httpx.HTTPStatusError as err: status = err.response.status_code if status == 401: raise UnauthorisedError("Request is unauthorised.") from err elif status == 403: raise ForbiddenError("Request is forbidden.") from err elif status == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise else: if not content_type: raise ValueError("content_type must be provided for text validation.") headers = {"Content-Type": content_type, "Accept": "text/turtle"} try: response = self.http_client.post( url=f"/rest/repositories/{repository_id}/validate/text", params=params, headers=headers, content=content, ) response.raise_for_status() return response.text except httpx.HTTPStatusError as err: status = err.response.status_code if status == 401: raise UnauthorisedError("Request is unauthorised.") from err elif status == 403: raise ForbiddenError("Request is forbidden.") from err elif status == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise class SecurityManagement: """GraphDB Security Management client.""" def __init__(self, http_client: httpx.Client): self._http_client = http_client @property def http_client(self): return self._http_client @property def enabled(self) -> bool: """Check if security is enabled. Returns: bool: True if security is enabled, False otherwise. Raises: ResponseFormatError: If the response format is invalid. InternalServerError: If the server returns an internal error. """ try: headers = {"Accept": "application/json"} response = self.http_client.get("/rest/security", headers=headers) response.raise_for_status() try: value = response.json() if not isinstance(value, bool): raise ResponseFormatError("Response is not a boolean.") return value except (ValueError, TypeError) as err: raise ResponseFormatError( f"Failed to parse GraphDB response: {err}" ) from err except httpx.HTTPStatusError as err: if err.response.status_code == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise @enabled.setter def enabled(self, value: bool): """Enable or disable security. Parameters: value: The value to set. Raises: TypeError: If the value is not a boolean. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. PreconditionFailedError: If the precondition is failed. """ if not isinstance(value, bool): raise TypeError("Value must be a boolean.") try: headers = {"Content-Type": "application/json", "Accept": "application/json"} response = self.http_client.post( "/rest/security", headers=headers, json=value ) response.raise_for_status() except httpx.HTTPStatusError as err: if err.response.status_code == 401: raise UnauthorisedError("Request is unauthorised.") from err elif err.response.status_code == 403: raise ForbiddenError("Request is forbidden.") from err elif err.response.status_code == 412: raise PreconditionFailedError("Precondition failed.") from err raise def get_free_access_details(self): """ Check if free access is enabled. Returns: FreeAccessSettings: The free access settings. Raises: ResponseFormatError: If the response format is invalid. InternalServerError: If the server returns an internal error. """ try: headers = {"Accept": "application/json"} response = self.http_client.get( "/rest/security/free-access", headers=headers ) response.raise_for_status() try: return FreeAccessSettings(**response.json()) except (ValueError, TypeError) as err: raise ResponseFormatError( f"Failed to parse GraphDB response: {err}" ) from err except httpx.HTTPStatusError as err: if err.response.status_code == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise def set_free_access_details(self, free_access_settings: FreeAccessSettings): """ Enable or disable free access. Parameters: free_access_settings: The free access settings. Raises: UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. PreconditionFailedError: If the precondition is failed. """ try: headers = {"Content-Type": "application/json", "Accept": "application/json"} response = self.http_client.post( "/rest/security/free-access", headers=headers, json=free_access_settings.as_dict(), ) response.raise_for_status() except httpx.HTTPStatusError as err: if err.response.status_code == 401: raise UnauthorisedError("Request is unauthorised.") from err elif err.response.status_code == 403: raise ForbiddenError("Request is forbidden.") from err elif err.response.status_code == 412: raise PreconditionFailedError("Precondition failed.") from err raise class UserManagement: """GraphDB User Management client.""" def __init__(self, http_client: httpx.Client): self._http_client = http_client @property def http_client(self): return self._http_client def list(self) -> list[User]: """ Get all users. Returns: A list of users. Raises: ResponseFormatError: If the response format is invalid. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. PreconditionFailedError: If the precondition is failed. InternalServerError: If the server returns an internal error. """ headers = {"Accept": "application/json"} try: response = self.http_client.get("/rest/security/users", headers=headers) response.raise_for_status() try: return [User(**user) for user in response.json()] except (ValueError, TypeError) as err: raise ResponseFormatError( f"Failed to parse GraphDB response: {err}" ) from err except httpx.HTTPStatusError as err: if err.response.status_code == 401: raise UnauthorisedError("Request is unauthorised.") from err elif err.response.status_code == 403: raise ForbiddenError("Request is forbidden.") from err elif err.response.status_code == 412: raise PreconditionFailedError("Precondition failed.") from err elif err.response.status_code == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise def get(self, username: str) -> User: """ Get a user. Parameters: username: The username of the user. Returns: User: The user. Raises: ResponseFormatError: If the response format is invalid. ForbiddenError: If the request is forbidden. NotFoundError: If the user is not found. InternalServerError: If the server returns an internal error. """ try: headers = {"Accept": "application/json"} response = self.http_client.get( f"/rest/security/users/{username}", headers=headers ) response.raise_for_status() try: return User(**response.json()) except (ValueError, TypeError) as err: raise ResponseFormatError( f"Failed to parse GraphDB response: {err}" ) from err except httpx.HTTPStatusError as err: if err.response.status_code == 403: raise ForbiddenError("Request is forbidden.") from err elif err.response.status_code == 404: raise NotFoundError("User not found.") from err elif err.response.status_code == 500: raise InternalServerError( f"Internal server error: {err.response.text}" ) from err raise def overwrite(self, username: str, user: User) -> None: """ Overwrite a user. Parameters: username: The username of the user. user: The user to overwrite. Raises: TypeError: if username is not a string or user is not an instance of User. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. NotFoundError: If the user is not found. """ if not isinstance(username, str): raise TypeError("Username must be a string.") if not isinstance(user, User): raise TypeError("User must be an instance of User.") try: headers = {"Content-Type": "application/json"} response = self.http_client.put( f"/rest/security/users/{username}", headers=headers, json=user.as_dict() ) response.raise_for_status() except httpx.HTTPStatusError as err: if err.response.status_code == 401: raise UnauthorisedError("Request is unauthorised.") from err elif err.response.status_code == 403: raise ForbiddenError("Request is forbidden.") from err elif err.response.status_code == 404: raise NotFoundError("User not found.") from err raise def create(self, username: str, user: UserCreate) -> None: """ Create a user. Parameters: username: The username of the user. user: The user to create. Raises: TypeError: if username is not a string or user is not an instance of UserCreate. BadRequestError: If the request is bad. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. """ if not isinstance(username, str): raise TypeError("Username must be a string.") if not isinstance(user, UserCreate): raise TypeError("User must be an instance of UserCreate.") try: headers = {"Content-Type": "application/json"} response = self.http_client.post( f"/rest/security/users/{username}", headers=headers, json=user.as_dict() ) response.raise_for_status() except httpx.HTTPStatusError as err: if err.response.status_code == 400: raise BadRequestError(f"Bad request. {err.response.text}") from err elif err.response.status_code == 401: raise UnauthorisedError("Request is unauthorised.") from err elif err.response.status_code == 403: raise ForbiddenError("Request is forbidden.") from err raise def delete(self, username: str) -> None: """ Delete a user. Parameters: username: The username of the user. Raises: TypeError: if username is not a string. BadRequestError: If the request is bad. UnauthorisedError: If the request is unauthorised. ForbiddenError: If the request is forbidden. NotFoundError: If the user is not found. """ if not isinstance(username, str): raise TypeError("Username must be a string.") try: response = self.http_client.delete(f"/rest/security/users/{username}") response.raise_for_status() except httpx.HTTPStatusError as err: if err.response.status_code == 400: raise BadRequestError(f"Bad request. {err.response.text}") from err elif err.response.status_code == 401: raise UnauthorisedError("Request is unauthorised.") from err elif err.response.status_code == 403: raise ForbiddenError("Request is forbidden.") from err elif err.response.status_code == 404: raise NotFoundError("User not found.") from err raise def update(self, username: str, user: UserUpdate | dict[str, t.Any]) -> None: """ Update a user. !!! Note Certain user fields are not updatable using PATCH. For example, at the time of writing, `grantedAuthorities` are not updatable, but `appSettings` are. Parameters: username: The username of the user. user: The user data to update, either as a UserUpdate instance or a dict. Raises: TypeError: if username is not a string or user is not an instance of UserUpdate or dict. ForbiddenError: If the request is forbidden. PreconditionFailedError: If the precondition is failed. """ if not isinstance(username, str): raise TypeError("Username must be a string.") if not isinstance(user, (UserUpdate, dict)): raise TypeError("User must be an instance of UserUpdate or dict.") if isinstance(user, UserUpdate): user_data = user.as_dict() else: user_data = user try: headers = {"Content-Type": "application/json"} response = self.http_client.patch( f"/rest/security/users/{username}", headers=headers, json=user_data ) response.raise_for_status() except httpx.HTTPStatusError as err: if err.response.status_code == 403: raise ForbiddenError("Request is forbidden.") from err elif err.response.status_code == 412: raise PreconditionFailedError("Precondition failed.") from err raise def custom_roles(self, username: str) -> t.List[str]: """ Retrieve custom roles associated with the user. Parameters: username: The username of the user. Returns: list[str]: The custom roles for the user. Raises: TypeError: if username is not a string. ForbiddenError: If the request is forbidden. NotFoundError: If the user is not found. InternalServerError: If the request fails. """ if not isinstance(username, str): raise TypeError("Username must be a string.") try: response = self.http_client.get( f"/rest/security/users/{username}/custom-roles" ) response.raise_for_status() return response.json() except httpx.HTTPStatusError as err: if err.response.status_code == 403: raise ForbiddenError("Request is forbidden.") from err elif err.response.status_code == 404: raise NotFoundError("User not found.") from err elif err.response.status_code == 500: raise InternalServerError("Internal server error.") from err raise class GraphDBClient(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: The base URL of the GraphDB server. auth: 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: Request timeout in seconds or an httpx.Timeout for fine-grained control (default: 30.0). kwargs: Additional keyword arguments to pass to the httpx.Client. """ 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) @property def cluster(self) -> ClusterGroupManagement: return self._cluster_group @property def monitoring(self) -> MonitoringManagement: return self._monitoring @property def recovery(self) -> RecoveryManagement: return self._recovery @property def repositories(self) -> RepositoryManager: """Server-level repository management operations (GraphDB-specific).""" return self._graphdb_repository_manager @property def graphdb_repositories(self) -> RepositoryManagement: return self._graphdb_repositories @property def security(self) -> SecurityManagement: return self._security @property def users(self) -> UserManagement: return self._users 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