datagraphs.Gateway

class documentation

class Gateway:

Constructor: Gateway(client, wait_time_ms)

Higher-level wrapper for deploying projects and bulk import/export of data.

Method	`__init__`	Initialise the Gateway.
Method	`clear_down`	Clear all data from all datasets in the project.
Method	`dump_data`	Export entity data from the project to JSON files on disk.
Method	`dump_project`	Export the project schema and dataset configurations to JSON files.
Method	`load_data`	Load data from JSON files into the DataGraphs project.
Method	`load_project`	Deploy the project schema and datasets to the API.
Constant	`UNKNOWN_PROJECT_NAME`	Undocumented
Property	`client`	The underlying `Client` instance.
Method	`_dump_datasets`	Dump the project datasets to JSON files.
Method	`_dump_schema`	Dump the project schema to a JSON file.
Method	`_ensure_no_duplicate_classes`	Validate that no duplicate class names are found across the provided datasets.
Method	`_get_project_name_from_entity`	Undocumented
Method	`_get_schema`	Undocumented
Method	`_load_data_for_class`	Load data for a specific class from a JSON file.
Method	`_load_data_for_dataset`	Load data for all non-base classes in a dataset.
Method	`_load_from_file`	Read a JSON file and PUT its contents into the project.
Method	`_map_data_project_urns`	Re-map URN project segments to match the current client project.
Method	`_persist_to_file`	Fetch entities of class_name from the API and write them to a JSON file.
Method	`_validate_datasets`	Validate that the local and API datasets match in terms of class names.
Method	`_verify_datasets_against_classlist`	Undocumented
Instance Variable	`_client`	Undocumented
Instance Variable	`_schema`	Undocumented
Instance Variable	`_wait_time_ms`	Undocumented

def __init__(self, client: DatagraphsClient, wait_time_ms: int = DatagraphsClient.DEFAULT_WAIT_TIME_MS): ¶

Initialise the Gateway.

Parameters
client:`DatagraphsClient`	A DataGraphs API client.
wait_time_ms:`int`	Delay in milliseconds between successive API calls.

def clear_down(self): ¶

Clear all data from all datasets in the project.

The datasets themselves are preserved.

def dump_data(self, to_dir_path: str | Path, class_name: str = Schema.ALL_CLASSES, include_date_fields: bool = False) -> dict: ¶

Export entity data from the project to JSON files on disk.

Each class is written to a separate <ClassName>.json file. Base classes are automatically skipped.

Parameters
to_dir_path:`str \| Path`	Target directory. Created automatically if it does not exist.
class_name:`str`	The class name to dump, or `Schema.ALL_CLASSES` for all.
include_date_fields:`bool`	Whether to include system date metadata.

Returns
`dict`	A dict with an `exported` count.

def dump_project(self, schema_path: str | Path, datasets_path: str | Path): ¶

Export the project schema and dataset configurations to JSON files.

Files are named {project_name}-v{version}-schema.json and {project_name}-v{version}-datasets.json.

Parameters
schema_path:`str \| Path`	Directory to write the schema JSON file.
datasets_path:`str \| Path`	Directory to write the datasets JSON file.

def load_data(self, class_name: str = Schema.ALL_CLASSES, from_dir_path: str | Path = '', file_path: str | Path = '') -> dict: ¶

Load data from JSON files into the DataGraphs project.

Base classes (those with subclasses) are automatically skipped when loading all classes.

Parameters
class_name:`str`	The class name to load, or `Schema.ALL_CLASSES` to load every non-base class found across all datasets.
from_dir_path:`str \| Path`	Directory containing `<ClassName>.json` files.
file_path:`str \| Path`	Explicit path to a single JSON file (used when loading a specific class_name).

Returns
`dict`	A dict with `loaded` and `skipped` counts.

def load_project(self, schema: Schema, datasets: list[Dataset], validation_mode: VALIDATION_MODE = VALIDATION_MODE.PROMPT): ¶

Deploy the project schema and datasets to the API.

Tears down existing datasets before applying the new schema and datasets.

Parameters
schema:`Schema`	The schema to deploy.
datasets:`list[Dataset]`	The datasets to deploy.
validation_mode:`VALIDATION_MODE`	How to handle mismatches between deployment and existing datasets. See `VALIDATION_MODE`.

Raises
`ValueError`	If duplicate class names are found across datasets.

UNKNOWN_PROJECT_NAME: str = ¶

Undocumented

Value

'__unknown__'

@property

client: DatagraphsClient = ¶

The underlying Client instance.

def _dump_datasets(self, datasets_path: str | Path, name_prefix: str): ¶

Dump the project datasets to JSON files.

def _dump_schema(self, schema_path: str | Path, name_prefix: str): ¶

Dump the project schema to a JSON file.

def _ensure_no_duplicate_classes(self, datasets: list[Dataset]): ¶

Validate that no duplicate class names are found across the provided datasets.

def _get_project_name_from_entity(self, entity: dict) -> str: ¶

Undocumented

def _get_schema(self) -> Schema: ¶

Undocumented

def _load_data_for_class(self, stats: dict, class_name: str, dataset_slug: str, from_dir_path: str | Path = '', file_path: str | Path = '') -> dict: ¶

Load data for a specific class from a JSON file.

def _load_data_for_dataset(self, stats: dict, dataset: Dataset, from_dir_path: str | Path) -> dict: ¶

Load data for all non-base classes in a dataset.

def _load_from_file(self, class_name: str, dataset_slug: str, from_dir_path: str | Path = '', file_path: str | Path = '') -> dict: ¶

Read a JSON file and PUT its contents into the project.

Returns:: A dict with loaded and skipped counts.
Raises:: FileNotFoundError: If file_path was explicitly provided but does not exist.

def _map_data_project_urns(self, data: list[dict]) -> list[dict]: ¶

Re-map URN project segments to match the current client project.

Args:: data: A list of entity dicts, each containing an id URN.
Returns:: The list with URNs re-mapped where necessary.
Raises:: ValueError: If an entity is missing a valid string id.

def _persist_to_file(self, class_name: str, to_dir_path: str | Path, include_date_fields: bool) -> int: ¶

Fetch entities of class_name from the API and write them to a JSON file.

Args:: class_name: The class name to fetch. to_dir_path: Target directory (must already exist).

def _validate_datasets(self, deployment_datasets: list[Dataset], existing_datasets: list[Dataset], validation_mode: VALIDATION_MODE): ¶

Validate that the local and API datasets match in terms of class names.

def _verify_datasets_against_classlist(self, datasets: list[Dataset], classes: list[str]): ¶

Undocumented

_client = ¶

Undocumented

_schema = ¶

Undocumented

_wait_time_ms = ¶

Undocumented