Skip to content

metadata_api_loader

metadata_api_loader ¤

DatapointAPI ¤

DatapointAPI(
    device_names: list[str],
    base_url: str,
    api_token: str,
    output_path: str = "data",
    required_uuid_list: list[str] | None = None,
    filter_enabled: bool = True,
    timeout: int = 30,
)

Loads datatron, device, and datapoint metadata from the Datadash REST API.

Authentication is bearer-token only — pass an externally obtained JWT token:

api = DatapointAPI(
    base_url="https://datadash.example.com",
    api_token="eyJhbGciOiJSUzI1NiJ9...",
    device_names=["Sensor A"],
)

The three core GET methods mirror the API hierarchy and can be called independently. get_all_uuids() aggregates them into the UUID lists required by AzureBlobParquetLoader.

:param device_names: Device names to collect datapoints for. :param base_url: API host, e.g. "https://datadash.example.com". :param api_token: JWT bearer token for Authorization: Bearer <token>. :param output_path: Directory to write per-device JSON exports. :param required_uuid_list: Optional allowlist; only matching UUIDs are kept. :param filter_enabled: When True, only datapoints with enabled=True are kept. :param timeout: Per-request timeout in seconds; prevents an unresponsive server from hanging the loader indefinitely.

get_datatrons ¤

get_datatrons() -> list[dict]

GET /api/datatrons — return all datatrons.

get_devices ¤

get_devices(datatron_id) -> list[dict]

GET /api/datatrons/{datatron_id}/devices — return all devices for a datatron.

get_datapoints ¤

get_datapoints(datatron_id, device_id) -> list[dict]

GET /api/datatrons/{datatron_id}/devices/{device_id}/data_points.

search_datatrons ¤

search_datatrons(
    query: str, fields: list[str] | None = None
) -> list[dict]

Filter datatrons whose fields contain query (case-insensitive).

search_devices ¤

search_devices(
    datatron_id, query: str, fields: list[str] | None = None
) -> list[dict]

Filter devices for a datatron whose fields contain query.

search_datapoints ¤

search_datapoints(
    datatron_id,
    device_id,
    query: str,
    fields: list[str] | None = None,
) -> list[dict]

Filter datapoints for a device whose fields contain query.

find_devices ¤

find_devices(
    query: str, fields: list[str] | None = None
) -> list[dict]

Search all devices across all datatrons.

Each result dict includes an extra datatron_id key.

find_datapoints ¤

find_datapoints(
    query: str, fields: list[str] | None = None
) -> list[dict]

Search all datapoints across all datatrons and devices.

Each result dict includes extra datatron_id and device_id keys.

get_all_uuids ¤

get_all_uuids() -> dict[str, list[str]]

Return {device_name: [uuid, ...]} — ready for the Azure parquet loader.

get_all_metadata ¤

get_all_metadata() -> dict[str, list[dict]]

Return {device_name: [record, ...]} with uuid/label/config columns.

display_dataframe ¤

display_dataframe(device_name: str | None = None) -> None

Log the metadata DataFrame for one device or all devices.