Metadata-Version: 2.4
Name: twinetic-api-client
Version: 0.2.2
Summary: HTTP client to access the API of a Twinetic EMS application or the Twinetic Hub server
License-Expression: BSD-3-Clause
License-File: LICENSE
Author: Richard Saeuberlich
Author-email: richard.saeuberlich@twinetic.de
Maintainer: Richard Saeuberlich
Maintainer-email: richard.saeuberlich@twinetic.de
Requires-Python: >=3.12,<4.0
Classifier: Development Status :: 4 - Beta
Classifier: Environment :: Web Environment
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Operating System :: OS Independent
Classifier: Topic :: Internet :: WWW/HTTP
Requires-Dist: httpx (>=0.28.1,<1.0.0)
Project-URL: Changelog, https://gitlab.com/twinetic-group/twinetic-api-client/-/blob/main/CHANGELOG.md
Project-URL: Homepage, https://twinetic.de/
Project-URL: Issues, https://gitlab.com/twinetic-group/twinetic-api-client/-/issues
Project-URL: Repository, https://gitlab.com/twinetic-group/twinetic-api-client
Description-Content-Type: text/markdown

# Twinetic API Client

A typed HTTP client for the [**Twinetic EMS**](https://twinetic.de/) 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/**](https://www.python-httpx.org/)

## Requirements

**Python:** `>=3.12,<4.0`

**Transitive Dependencies:**

| Package | Description |
| --- | --- |
| `anyio` | High-level concurrency and networking framework on top of asyncio or Trio |
| `certifi` | Python package for providing Mozilla's CA Bundle. |
| `h11` | A pure-Python, bring-your-own-I/O implementation of HTTP/1.1 |
| `httpcore` | A minimal low-level HTTP client. |
| `httpx` | The next generation HTTP client. |
| `idna` | Internationalized Domain Names in Applications (IDNA) |

## Installation

```shell
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.

```python
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.

```python
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.

```python
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.

```python
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.

```python
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:

```python
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`.

```python
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:

```python
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.

```python
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()
```

