intellistream-datahub-sdk

Python bindings for the IntelliStream DataHub SDK — a client for the DataHub REST API, implemented in Rust and exposed to Python with PyO3.

pip install intellistream-datahub-sdk

from datetime import datetime, timezone

import intellistream_datahub_sdk as dh

client = dh.DataHubClient.from_envfile(".env")

# Create a time series, then ingest a datapoint into it.
client.timeseries.create([dh.TimeSeries(external_id="sensor_1", name="Sensor 1")])
client.timeseries.insert_from_lists(
    timestamps=[datetime.now(timezone.utc)],
    values=[21.5],
    ts="sensor_1",
)

# Find the alarms.
alarms = client.events.filter(type="ALARM", limit=100)

# The same criteria, kept as an object, so one definition can be filtered and searched with.
ALARMS = dh.EventFilter(type="ALARM")
recent = client.events.filter(filter=ALARMS, limit=100, sort_by="eventTime", sort_order="desc")
matches = client.events.search("bearing", filter=ALARMS)

Every filter() takes either the criteria as keywords or a prepared filter= object — passing both is a TypeError. limit, sort_by, sort_order and cursor are always arguments of the call rather than fields of the filter, so a stored filter carries no paging state into its next use.

Both a synchronous and an asynchronous client are available — DataHubClient and AsyncDataHubClient. The async one exposes the same services with awaitable methods.

Anywhere an entity is identified you may pass its external id, its numeric id, an IdCollection, or the entity itself.

What it covers

Configuration

The client reads its configuration from a .env file or the environment:

• BASE_URL — the backend root, e.g. https://datahub.example.com
• either TOKEN — a bearer token used as-is
• or CLIENT_ID, CLIENT_SECRET and TOKEN_URI — an OAuth2 client-credentials exchange

Type hints

The package ships py.typed and a full .pyi stub, so mypy and IDE completion work out of the box.

License

Apache License 2.0 — see LICENSE.

Source: https://github.com/IntelliStream-DataHub/dataplatform-rust-sdk