datagraphs.Client

class documentation

class Client:

Constructor: Client(project_name, api_key, client_id, client_secret, ...)

Low-level HTTP client for the DataGraphs REST API.

Handles authentication, pagination, and batched writes.

Method	`__init__`	Initialise the DataGraphs client.
Method	`apply_datasets`	Create or update datasets so they match the supplied list.
Method	`apply_schema`	Apply a schema to the project, replacing the currently active domain model.
Method	`clear_dataset`	Delete all data from a dataset, keeping the dataset itself intact.
Method	`create_dataset`	Create a new dataset.
Method	`delete`	Delete a single entity by class and ID.
Method	`drop_dataset`	Drop a dataset and all its data entirely.
Method	`get`	Retrieve all entities of a given type.
Method	`get_datasets`	Retrieve all datasets in the project.
Method	`get_schema`	Retrieve the active schema for the project.
Method	`put`	Load entities into a dataset.
Method	`query`	Query the API with filters, facets, sorting, and pagination.
Method	`set_wait_time`	Set the wait time between paginated requests.
Method	`status`	Check the API service status.
Method	`tear_down`	Remove all datasets and their data from the project.
Method	`update_dataset`	Update an existing dataset.
Constant	`AUTH_URL_SUFFIX`	Undocumented
Constant	`DATASETS_TIMEOUT_MS`	Undocumented
Constant	`DEFAULT_BATCH_SIZE`	Undocumented
Constant	`DEFAULT_DATASETS_PAGE_SIZE`	Undocumented
Constant	`DEFAULT_FACET_SIZE`	Undocumented
Constant	`DEFAULT_WAIT_TIME_MS`	Undocumented
Constant	`HTTP_CREATED`	Undocumented
Constant	`HTTP_FORBIDDEN`	Undocumented
Constant	`HTTP_GATEWAY_TIMEOUT`	Undocumented
Constant	`HTTP_NO_CONTENT`	Undocumented
Constant	`HTTP_OK`	Undocumented
Constant	`HTTP_UNAUTHORIZED`	Undocumented
Constant	`MAX_AUTH_RETRIES`	Undocumented
Constant	`PROD_URL`	Undocumented
Instance Variable	`project_name`	Undocumented
Property	`wait_time_ms`	The current wait time in milliseconds between paginated requests.
Static Method	`_build_query`	Build a query string from key-value pairs.
Method	`_assert_datasets_applied`	Undocumented
Method	`_cache_buster`	Return a cache-busting timestamp value.
Method	`_get_auth_token`	Undocumented
Method	`_get_data_url`	Undocumented
Method	`_get_headers`	Undocumented
Method	`_get_query_url`	Undocumented
Method	`_has_oauth_credentials`	Undocumented
Method	`_request`	Execute an HTTP request with automatic auth retry.
Instance Variable	`_api_key`	Undocumented
Instance Variable	`_auth_token`	Undocumented
Instance Variable	`_batch_size`	Undocumented
Instance Variable	`_client_id`	Undocumented
Instance Variable	`_client_secret`	Undocumented
Instance Variable	`_http_client`	Undocumented
Instance Variable	`_service_url`	Undocumented
Instance Variable	`_wait_time_ms`	Undocumented
Property	`_base_url`	Undocumented

def __init__(self, project_name: str, api_key: str, client_id: str = '', client_secret: str = '', batch_size: int = DEFAULT_BATCH_SIZE, service_url: str = PROD_URL): ¶

Initialise the DataGraphs client.

Parameters
project_name:`str`	Name of the project.
api_key:`str`	API key for authentication.
client_id:`str`	OAuth client ID (required for write operations).
client_secret:`str`	OAuth client secret (required for write operations).
batch_size:`int`	Number of items to process in each batch.
service_url:`str`	Base URL for the API service.

def apply_datasets(self, datasets: list[Dataset], timeout_ms: int = DATASETS_TIMEOUT_MS): ¶

Create or update datasets so they match the supplied list.

New datasets are created; existing datasets with changes are updated. Waits for confirmation that all datasets have been applied.

Parameters
datasets:`list[Dataset]`	Datasets to apply.
timeout_ms:`int`	Maximum time in milliseconds to wait for the API to confirm all datasets are applied.

Raises
`DatagraphsError`	If datasets are not applied within the timeout.

def apply_schema(self, schema: DatagraphsSchema): ¶

Apply a schema to the project, replacing the currently active domain model.

Parameters
schema:`DatagraphsSchema`	The schema to apply.

def clear_dataset(self, dataset_slug: str): ¶

Delete all data from a dataset, keeping the dataset itself intact.

Parameters
dataset_slug:`str`	The slug of the dataset to clear.

def create_dataset(self, dataset: Dataset): ¶

Create a new dataset.

Parameters
dataset:`Dataset`	The dataset to create.

def delete(self, class_name: str, entity_id: str): ¶

Delete a single entity by class and ID.

Parameters
class_name:`str`	The entity class.
entity_id:`str`	The entity identifier.

def drop_dataset(self, dataset_slug: str): ¶

Drop a dataset and all its data entirely.

Parameters
dataset_slug:`str`	The slug of the dataset to drop.

def get(self, class_name: str, lang: str = 'all', include_date_fields: bool = False) -> list[dict[str, Any]]: ¶

Retrieve all entities of a given type.

Automatically paginates through all results.

Parameters
class_name:`str`	The entity class to fetch.
lang:`str`	Language code for results (default `'all'`).
include_date_fields:`bool`	Whether to include system date metadata.

Returns
`list[dict[str, Any]]`	A list of entity dicts.

def get_datasets(self) -> list[Dataset]: ¶

Retrieve all datasets in the project.

Returns
`list[Dataset]`	A list of `Dataset` objects.

def get_schema(self) -> DatagraphsSchema: ¶

Retrieve the active schema for the project.

Returns
`DatagraphsSchema`	The current project `Schema`.

def put(self, dataset: str, data: dict[str, Any] | list[dict[str, Any]]) -> int: ¶

Load entities into a dataset.

Automatically batches large payloads according to the configured batch_size.

Parameters
dataset:`str`	Target dataset slug.
data:`dict[str, Any] \| list[dict[str, Any]]`	A single entity dict or a list of entity dicts.

Returns
`int`	The number of entities loaded.

def query(self, dataset: str = '_all', q: str = '', filters: str = '', facets: str = '', facet_size: int = -1, date_facets: str = '', fields: str = '', embed: str = '', sort: str = '', ids: str = '', lang: str = 'all', page_no: int = -1, page_size: int = -1, previous_page_token: str = '', next_page_token: str = '', include_date_fields: bool = False) -> list[dict[str, Any]] | tuple[list[dict[str, Any]], list[dict[str, Any]]]: ¶

Query the API with filters, facets, sorting, and pagination.

Parameters
dataset:`str`	Dataset slug to query (default `'_all'`).
q:`str`	Free-text search query string.
filters:`str`	Filter expression (e.g. `'type:Person'`).
facets:`str`	Comma-separated facet field names.
facet_size:`int`	Number of facet values to return (default `10`).
date_facets:`str`	Date facet specification.
fields:`str`	Comma-separated field names to include in results.
embed:`str`	Embedding depth for related entities.
sort:`str`	Sort expression (e.g. `'label:asc'`).
ids:`str`	Comma-separated entity IDs to fetch directly.
lang:`str`	Language code for results (default `'all'`).
page_no:`int`	Page number for offset-based pagination.
page_size:`int`	Number of results per page.
previous_page_token:`str`	Token for cursor-based backward pagination.
next_page_token:`str`	Token for cursor-based forward pagination.
include_date_fields:`bool`	Whether to include system date metadata.

Returns
`list[dict[str, Any]] \| tuple[list[dict[str, Any]], list[dict[str, Any]]]`	A list of result dicts, or a `(results, facets)` tuple when facets are requested.

def set_wait_time(self, wait_time_ms: int): ¶

Set the wait time between paginated requests.

Parameters
wait_time_ms:`int`	Wait time in milliseconds.

def status(self) -> str: ¶

Check the API service status.

Returns
`str`	The API status string, or `'unknown'` if unavailable.

def tear_down(self, drop_datasets: bool = True): ¶

Remove all datasets and their data from the project.

Parameters
drop_datasets:`bool`	If `True`, drops each dataset entirely. If `False`, only clears the data from each dataset.

def update_dataset(self, dataset: Dataset): ¶

Update an existing dataset.

Parameters
dataset:`Dataset`	The dataset to update (matched by slug).

AUTH_URL_SUFFIX: str = ¶

Undocumented

Value

'oauth/token'

DATASETS_TIMEOUT_MS: int = ¶

Undocumented

Value

DEFAULT_BATCH_SIZE: int = ¶

Undocumented

Value

DEFAULT_DATASETS_PAGE_SIZE: int = ¶

Undocumented

Value

DEFAULT_FACET_SIZE: int = ¶

Undocumented

Value

DEFAULT_WAIT_TIME_MS: int = ¶

Undocumented

Value

HTTP_CREATED: int = ¶

Undocumented

Value

HTTP_FORBIDDEN: int = ¶

Undocumented

Value

HTTP_GATEWAY_TIMEOUT: int = ¶

Undocumented

Value

HTTP_NO_CONTENT: int = ¶

Undocumented

Value

HTTP_OK: int = ¶

Undocumented

Value

HTTP_UNAUTHORIZED: int = ¶

Undocumented

Value

MAX_AUTH_RETRIES: int = ¶

Undocumented

Value

PROD_URL: str = ¶

Undocumented

Value

'https://api.datagraphs.io/'

project_name = ¶

Undocumented

@property

wait_time_ms: int = ¶

The current wait time in milliseconds between paginated requests.

@staticmethod

def _build_query(params: list[tuple[str, str]]) -> str: ¶

Build a query string from key-value pairs.

def _assert_datasets_applied(self, datasets: list[Dataset], timeout_ms: int): ¶

Undocumented

def _cache_buster(self) -> str: ¶

Return a cache-busting timestamp value.

def _get_auth_token(self, force_refresh=False) -> str: ¶

Undocumented

def _get_data_url(self, class_name: str, page_size: int, lang: str, include_date_fields: bool, next_page_token: str = '', page_no: int = -1) -> str: ¶

Undocumented

def _get_headers(self, lang: str = 'all') -> dict[str, str]: ¶

Undocumented

def _get_query_url(self, dataset: str = '_all', q: str = '', filters: str = '', facets: str = '', facet_size: int = -1, date_facets: str = '', fields: str = '', embed: str = '', sort: str = '', ids: str = '', lang: str = 'all', page_no: int = -1, page_size: int = -1, previous_page_token: str = '', next_page_token: str = '', include_date_fields: bool = False) -> str: ¶

Undocumented

def _has_oauth_credentials(self) -> bool: ¶

Undocumented

def _request(self, method: HTTP, url: str, _retry_count: int = 0, **kwargs) -> dict[str, Any] | None: ¶

Execute an HTTP request with automatic auth retry.

Parameters
method:`HTTP`	The HTTP method to use.
url:`str`	The request URL.
_retry_count:`int`	Internal retry counter (do not set externally).
**kwargs	Additional arguments forwarded to the HTTP client.

Returns
`dict[str, Any] \| None`	Parsed JSON response for GET requests, or `None` for mutating requests.

Raises
`AuthenticationError`	If authentication fails after max retries.
`DatagraphsError`	If the request fails for any other reason.

_api_key = ¶

Undocumented

_auth_token = ¶

Undocumented

_batch_size = ¶

Undocumented

_client_id = ¶

Undocumented

_client_secret = ¶

Undocumented

_http_client = ¶

Undocumented

_service_url = ¶

Undocumented

_wait_time_ms = ¶

Undocumented

@property

_base_url: str = ¶

Undocumented