Python client

Verity maintains an open-source Python package, datahub-api-connector, that wraps the API: it acquires and renews the token, keeps a connection pool, retries what is worth retrying, and turns keyword arguments into query parameters.

It is a thin layer — as its README puts it, no magic. Requests and responses stay plain HTTP, so everything in Querying the API applies unchanged.
Package datahub-api-connector
Version documented here 1.6
Requires Python 3.10+
Licence MIT

Golden rule

This package works only from Insights 7.0 onward.

Version 7.0 (1 July 2025) moved authentication to Keycloak, and the connector speaks only that. For an older platform, the previous opinum-api-connector package is the one to use — it is not maintained any more, so this is a reason to upgrade rather than a supported path.

Install

pip install datahub-api-connector

Configure

Credentials are read from environment variables, not from constructor arguments.

Variable Required Holds
DATAHUB_USERNAME Yes The Insights user the calls act as.
DATAHUB_PASSWORD Yes Its password.
DATAHUB_CLIENT_ID Yes The application client id.
DATAHUB_CLIENT_SECRET Yes The matching secret.
DATAHUB_SCOPE No Defaults to datahub-api. Use datahub-api push-data to push.
DATAHUB_API_URL No Another API host than the European SaaS one.
DATAHUB_AUTH_URL No Another authentication host.
DATAHUB_PUSH_URL No Another push host.

See Connect to the API for how to obtain the four credentials, and Scopes for what each scope opens.

Tip

Pass an environment dictionary to the constructor to read the settings from somewhere other than os.environ — a secret manager, or a test fixture.

A first call

from datahub_api_connector import ApiConnector

with ApiConnector(account_id=1234) as api:
    sources = api.get('sources', DisplayLevel='Light').json()
    print(f"Found {len(sources)} sources")

Every keyword argument becomes a query parameter, with one exception: data is the request body. So api.get('sources', DisplayLevel='Light', SiteId=42) produces GET /sources?DisplayLevel=Light&SiteId=42.

The methods return the raw requests response — call .json() yourself.

Method Use
get, post, patch, put, delete The API on api.opinum.com.
push_data(body, operation_id=None, operation_timeout_sec=None) The push service, in the standard format.
push_dataframe_data(df, **kwargs) Same, from a pandas DataFrame with date and value columns.
send_file_to_storage(filename, file_io, mime_type) Upload to storage.
close() Release the connection pool. The context manager does it for you.

Accounts

account_id selects the tenant, and it is sent on every token request, renewals included.

api = ApiConnector(account_id=1234)
api.account_id = 5678          # takes effect on the very next call
print(api.token_account_id)    # which account the current token really claims

Golden rule

If the user has access to several tenants, always set account_id.

Without it the connector uses the last tenant that user touched — which is a property of the past, not of your code. Reading and writing the wrong account is a failure that produces no error at all, only wrong data in a real place.

Pushing data

api.push_data(
    [{"variableId": 7174702,
      "data": [{"date": "2026-08-28T10:00:00", "value": 12.4}]}],
    operation_id="nightly-2026-08-28",
)

The payload is the standard format. Passing your own operation_id is what ties a webhook callback — or a support ticket — back to the run that produced it.

Important

Pushing requires the push-data scope. Set DATAHUB_SCOPE="datahub-api push-data", or the push is rejected with a 401 while your reads keep working.

Retries

This is the part worth understanding, because the connector already does what most integrations hand-roll badly.

How a failed call is handled

Retries are on by default since 1.6: three extra attempts, with the wait doubling each time (seconds_between_retries, capped at 60 seconds) plus a jitter so threads sharing an instance stop retrying in lockstep. A Retry-After header always wins over the computed wait.

Parameter Default Role
retries_when_connection_failure 3 (max 5) Extra attempts. 0 restores the pre-1.6 single-attempt behaviour.
retry_on_status 408, 425, 429, 500, 502, 503, 504 Statuses retried rather than raised. None disables.
retry_unsafe_methods False Apply those statuses to POST/PUT/PATCH/DELETE too.
seconds_between_retries 5 The first wait.
request_timeout 10 Seconds, on every request including the token.
pool_size 32 Connection pool. Must be ≥ the number of threads sharing the instance.

Two behaviours are handled apart and cost nothing from the attempt budget:

  • A 401 on a token that still looked valid — revoked, clock skew, account switched server-side — renews the token and replays the call once.
  • A rejected credential or scope is raised straight away. So is any other 4xx: a second attempt would fail identically.

Golden rule

Leave retry_unsafe_methods off unless the write is safe to replay.

A retried write that the server had already applied before failing duplicates the change. Pushing the same data points at the same timestamps overwrites, so it is safe; a POST /sources replayed after a timeout can create a second source. The query-by-body POST /data only reads, and is a fair exception.

Parallel calls

multi_thread_request_on_path splits one long parameter list into several calls and runs them in parallel — the answer to the URL length limit described in Singular and plural parameters.

from datahub_api_connector import ApiConnector, multi_thread_request_on_path

with ApiConnector(account_id=1234) as api:
    for response in multi_thread_request_on_path(
            api.get, 'data',
            split_parameter='VariableIds',
            max_parameter_entities=80,     # ids per call — keeps the URL short enough
            max_futures=8,                 # calls in flight at once
            workers=16,                    # threads; must not exceed pool_size
            VariableIds=variable_ids,
            DisplayLevel='ValueVariableDateSource',
            Granularity='Day', Aggregation='SUM',
            From='2026-08-01T00:00:00Z', To='2026-08-31T23:59:59Z',
            IncludeToBoundary='true'):
        rows.extend(response.json())

It is a generator: results arrive as calls complete. When one call of a group fails, the ones that already succeeded are yielded first and the failure is raised afterwards, so work the API has already done is not thrown away. Pass raise_on_error=False to collect everything that worked and have the rest logged only.

Tip

response_callback lets you hand back something other than the raw response — lambda r: r.json(), for instance — so the generator yields the data directly.

Counting without downloading

IncludeItemsCount=True adds an x-total-count response header on the endpoints that support it:

response = api.get('sources', DisplayLevel='Light', IncludeItemsCount=True)
total = response.headers.get('x-total-count', 'unknown')

Reconcile what you asked for

Golden rule

Compare the ids you requested with the ids you received.

A variable with no data in the window is absent from the response rather than returned with a zero. Iterating only over the returned rows makes a silent meter look exactly like a healthy one.

returned = {row["variableId"] for row in rows}
missing = set(variable_ids) - returned

Logging

The package configures only its own logger — log_level="INFO" by default — and leaves the root logger as your application set it. Retry and failure logs name the method, the URL and the start of the response body, which is where the API states the actual cause.

DOCS 2026.08 REVIEWED 2026-08-28 API 1.7 Documentation changelog →