Twinetic API Client

A typed HTTP client for the Twinetic EMS REST API. Handles JWT authentication and token refresh transparently, and gives per-endpoint methods so you don't have to build URLs or parse pagination by hand.

Base of this client is the excellent HTTPX library: https://www.python-httpx.org/

Requirements

Python: >=3.12,<4.0

Transitive Dependencies:

Installation

pip install twinetic-api-client

Authentication

The client authenticates with a refresh token — your personal access token (PAT), which you generate on your Twinetic EMS instance. The client exchanges it for short-lived access tokens automatically and re-mints them when they expire; you never handle access tokens yourself.

from twinetic.clients.ems.twinetic import TwineticClient

with TwineticClient(base_url="https://my-ems-instance.twinetic.de", refresh_token="<YOUR-GENERATED-REFRESH-TOKEN>") as client:
    meters = client.all_meters()

Use the client as a context manager (with) so the underlying connection is closed cleanly, or call client.close() yourself.

Reading data

Every list endpoint offers four access patterns, so you can choose how much data to fetch and how.

A single element by id. Fetch one record directly. Meter devices are looked up by UUID; measurements, units, prefixes, PQMs, and datapoints by their integer id.

from typing import Any
from twinetic.clients.ems.twinetic import TwineticClient

client = TwineticClient(base_url="https://my-ems-instance.twinetic.de", refresh_token="<YOUR-GENERATED-REFRESH-TOKEN>")

try:
    one_meter: dict[str, Any] = client.meter("fd7f55b5-4c8d-4b50-8cbb-6ceb95d04f5f")    # by UUID
    one_measurement: dict[str, Any] = client.single_measurement(40)                     # by id
finally:
    client.close()

A nonexistent id raises TwineticAPIError (404).

One page, with pagination info. Returns a RestfulResponse carrying count, next, previous, and results — reach for this when you want to page manually and need to know how many pages remain.

from twinetic.clients.dto import RestfulResponse
from twinetic.clients.ems.twinetic import TwineticClient

client = TwineticClient(base_url="https://my-ems-instance.twinetic.de", refresh_token="<YOUR-GENERATED-REFRESH-TOKEN>")

try:
    page: RestfulResponse = client.meters(page=1)
    print(page.count)      # total count of all meter units across all pages
    print(page.next)       # URL of the next page, or None at the end
    print(page.previous)   # URL of the previous page, or None at the beginning
    print(page.results)    # The entities of this page
finally:
    client.close()

All rows, eager. Fetches every page up front and returns them as one flat list. Every row is held in memory at once. Use this when you want the complete set and intend to keep it around.

from twinetic.clients.ems.twinetic import TwineticClient

client = TwineticClient(base_url="https://my-ems-instance.twinetic.de", refresh_token="<YOUR-GENERATED-REFRESH-TOKEN>")

try:
    for meter in client.all_meters():
        print(meter["name"])
finally:
    client.close()

All rows, lazy. Yields rows one page at a time, fetching each page only as you reach it — and only one page is held in memory at a time. Stop iterating early and the remaining pages are never fetched. Use this to stream large result sets or to search without pulling everything.

from twinetic.clients.ems.twinetic import TwineticClient

client = TwineticClient(base_url="https://my-ems-instance.twinetic.de", refresh_token="<YOUR-GENERATED-REFRESH-TOKEN>")

try:
    for meter in client.all_meters_lazy():
        if meter["name"] == "My special meter unit name":
            break   # only the pages needed to reach this row were fetched
finally:
    client.close()

The single-page method returns a RestfulResponse; the eager and lazy methods return the rows directly (a list[dict[str, Any]] and an iterator of dict[str, Any]). Rows are plain dictionaries — the same shape the API returns.

Choosing an API version

The client targets a default REST version set at construction (default_version, RESTVersion.V1 by default). Any single call can override it:

from typing import Any
from twinetic.clients.ems.twinetic import TwineticClient
from twinetic.clients.constants import RESTVersion

client = TwineticClient(base_url="https://my-ems-instance.twinetic.de", refresh_token="<YOUR-GENERATED-REFRESH-TOKEN>", default_version=RESTVersion.V1)

try:
    meters_v1: list[dict[str, Any]] = client.all_meters()                               # uses the default (v1) that was given during instantiation of the client
    meters_v2: list[dict[str, Any]] = client.all_meters(rest_version=RESTVersion.V2)    # this call uses v2
finally:
    client.close()

Asynchronous Clients

Every client has an async twin, e.g. TwineticClient & AsyncTwineticClient. The asynchronous variants expose the same methods, just awaited. Eager and single-element methods are awaited; the _lazy iterators are consumed with async for.

import asyncio

from twinetic.clients.ems.twinetic import AsyncTwineticClient


async def main() -> None:
    async with AsyncTwineticClient(base_url="https://my-ems-instance.twinetic.de", refresh_token="<YOUR-GENERATED-REFRESH-TOKEN>") as client:
        # eager and single-element: await
        for meter in await client.all_meters():
            print(meter["name"])

        one = await client.single_meter("d244eada-7e97-40cb-aeb2-e6183be7fe06")
        print(one["name"])

        # lazy: async for
        async for meter in client.all_meters_lazy():
            print(meter["name"])


asyncio.run(main())

Use the async client as an async context manager (async with) so the connection closes cleanly, or call await client.aclose() yourself.

Error handling

Any 4xx/5xx response raises TwineticAPIError, which carries the status code and the parsed response body:

from typing import Any
from twinetic.clients.ems.twinetic import TwineticClient
from twinetic.clients.exception import TwineticAPIError

client = TwineticClient(base_url="https://my-ems-instance.twinetic.de", refresh_token="<YOUR-GENERATED-REFRESH-TOKEN>")

try:
    # NOTE: 99th page does NOT exist in this example, therefore this call will raise an error
    client.units(99)
except TwineticAPIError as exc:
    print(exc.status_code)
    print(exc.body)
finally:
    client.close()

An expired or revoked refresh token raises RefreshTokenInvalidError — generate a new JWT refresh token on your Twinetic EMS instance and construct the client from new with it.

from twinetic.clients.ems.twinetic import TwineticClient
from twinetic.clients.exception import RefreshTokenInvalidError

client = TwineticClient(base_url="https://my-ems-instance.twinetic.de", refresh_token="<YOUR-GENERATED-REFRESH-TOKEN>")

try:
    client.all_meters()
except RefreshTokenInvalidError:
    print("Refresh token no longer valid — generate a new PAT and reconstruct the client.")
finally:
    client.close()