Metadata-Version: 2.4
Name: flespi-client
Version: 0.1.0
Summary: Async Python SDK for the flespi IoT/telematics REST API
Author-email: nashimanovskii <nashimanovskii@gmail.com>
License-Expression: Apache-2.0
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.13
Requires-Dist: aiohttp~=3.14.1
Requires-Dist: prometheus-client~=0.24.1
Requires-Dist: pydantic~=2.13.4
Requires-Dist: redis~=8.0.1
Description-Content-Type: text/markdown

# flespi-client

Async Python SDK for the [flespi](https://flespi.io) IoT/telematics REST API — GPS tracking, fleet management, geofencing, trip/mileage analytics, real-time alerting, multi-tenant token scoping.

Every resource path, selector, and request/response body is validated with Pydantic **before** a request goes out — a typo'd field name, a malformed selector, or a body that doesn't match flespi's real schema fails in Python, not after a round trip.

- **Fluent, typed resource tree** mirroring flespi's own REST structure: `client.gw.devices(selector).telemetry()`, `client.platform.tokens()`, `client.gw.calcs(selector).devices(selector)`, ...
- **Selectors are a typed AST**, not raw strings (`flespi_client/core/selectors/expr.py`) — build them with Python operators (`FlespiField("speed", ">", 90) & FlespiField("blocked", "==", 0)`), not string concatenation.
- **Automatic token refresh** on 401, **retry with backoff** on transient 5xx/connection errors, **bulk-POST chunking** against flespi's real per-endpoint item caps, an optional **distributed lock** for concurrent writers, and **Prometheus metrics** out of the box.
- **Generic `metadata`/`configuration` fields** — every opaque JSON blob flespi accepts can be parametrized with your own Pydantic model instead of an untyped dict.

## Table of contents

- [Install](#install)
- [Quick start](#quick-start)
- [Business use cases](#business-use-cases)
  - [1. Fleet monitoring dashboard](#1-fleet-monitoring-dashboard)
  - [2. Bulk device onboarding](#2-bulk-device-onboarding)
  - [3. Geofence zone alerts](#3-geofence-zone-alerts)
  - [4. Trip & mileage analytics via calculators](#4-trip--mileage-analytics-via-calculators)
  - [5. Multi-tenant SaaS: scoped tokens per customer](#5-multi-tenant-saas-scoped-tokens-per-customer)
  - [6. Real-time alerting via webhooks](#6-real-time-alerting-via-webhooks)
- [Architecture](#architecture)
- [Metrics](#metrics)
- [Development](#development)

## Install

```bash
uv add flespi-client
```

## Quick start

```python
import asyncio

from flespi_client.core.client import FlespiClient
from flespi_client.core.selectors import PathSelector


async def main() -> None:
    async with FlespiClient(token="YOUR_FLESPI_TOKEN") as client:
        devices = await client.gw.devices(PathSelector()).get()
        print(devices)


asyncio.run(main())
```

Login with username/password/realm instead of a static token:

```python
async with FlespiClient(
    username="user@example.com",
    password="...",
    realm_id=1,
) as client:
    ...
```

## Business use cases

The snippets below are complete enough to adapt directly — every field name and schema is taken from this SDK's actual `flespi_client/schemas/` package, not simplified for the README.

### 1. Fleet monitoring dashboard

List every vehicle and pull its latest telemetry — the core loop behind any live map/dashboard.

```python
from flespi_client.core.selectors import PathSelector
from flespi_client.core.selectors.expr import FlespiField

async with FlespiClient(token=token) as client:
    fleet = await client.gw.devices(PathSelector()).get()

    for device in fleet:
        selector = PathSelector(FlespiField("id", "==", device.id))
        telemetry = await client.gw.devices(selector).telemetry(PathSelector()).get()
        position = (telemetry[0].telemetry or {}).get("position") if telemetry else None
        print(f"{device.name}: {position}")
```

`device.metadata`/`device.configuration` default to plain dicts — parametrize `client.gw.devices[MyMetadata, MyConfig]` if your fleet metadata (owner, depot, contract ID, ...) has a fixed shape you want validated and typed instead of an arbitrary JSON blob (see the generic-resources example under [Architecture](#architecture)).

### 2. Bulk device onboarding

Provisioning a batch of new trackers is one call — bodies are validated per-item, and a batch larger than flespi's real per-request cap (`MAX_DEVICES_PER_REQUEST`) is split into sequential requests automatically:

```python
from flespi_client.schemas.devices.base import DeviceSchema

new_devices = [
    DeviceSchema(name=f"tracker-{ident}", device_type_id=1, configuration={"ident": ident})
    for ident in imeis  # e.g. 5,000 IMEIs from a hardware shipment
]

created = await client.gw.devices().post(new_devices, chunk_size=500)
print(f"onboarded {len(created)} devices")
```

If a chunk fails after earlier ones already succeeded, `.post()` raises `ChunkedPostError` carrying `.partial_results` (what already exists server-side) and `.failed_chunk_index` — so a failed batch never leaves you guessing what went through.

### 3. Geofence zone alerts

Define a zone once, then check whether a live GPS fix falls inside it — the basis of arrival/departure and dwell-time alerting.

```python
import json

from flespi_client.schemas.geofences.base import CircleGeometrySchema, GeofenceSchema
from flespi_client.schemas.selectors.geofences import CoordinatesSchema

warehouse = (
    await client.gw.geofences().post(
        GeofenceSchema(
            name="warehouse-1",
            geometry=CircleGeometrySchema(
                center=CoordinatesSchema(lat=55.751244, lon=37.618423),
                radius=0.5,  # km
            ),
        )
    )
)[0]

# hittest: which geofence(s) contain a given point, right now
hit = await client.gw.geofences(PathSelector()).hittest().get(
    params={"data": json.dumps({"lat": 55.7520, "lon": 37.6190})}
)
print(f"vehicle is inside: {[zone.name for zone in hit]}")
```

For continuous monitoring rather than one-off checks, attach a `geofence`-type interval selector to a calculator (below) instead of polling `hittest` yourself.

### 4. Trip & mileage analytics via calculators

flespi's calculators turn a raw message stream into intervals (trips, stops, idle periods, ...) server-side — no need to reprocess telemetry yourself. This defines "moving" as speed above a threshold and sums `mileage()` per trip:

```python
from flespi_client.schemas.calcs.base import CalcSchema
from flespi_client.schemas.counters.base import ExpressionCounterSchema
from flespi_client.schemas.counters.enums import CounterExpressionMethodEnum
from flespi_client.schemas.selectors.base import ExpressionSelectorSchema

calc = (
    await client.gw.calcs().post(
        CalcSchema(
            name="daily-mileage",
            selectors=[
                ExpressionSelectorSchema(expression="$position.speed>3", name="moving")
            ],
            counters=[
                ExpressionCounterSchema(
                    name="distance_km",
                    expression="mileage()",
                    method=CounterExpressionMethodEnum.summary,
                )
            ],
        )
    )
)[0]

# Assign the fleet to it — POST body here is optional per-assignment config
# (enabled/time_begin/time_end); the target device(s) are the *selector*.
calc_selector = PathSelector(FlespiField("id", "==", calc.id))
for device in fleet:
    device_selector = PathSelector(FlespiField("id", "==", device.id))
    await client.gw.calcs(calc_selector).devices(device_selector).post({})
```

`mileage()`/`Fn.mileage()` and every other flespi expression function are available as typed builders in `flespi_client.core.selectors.functions.Fn`, so counter/selector expressions can be composed the same way path selectors are, instead of hand-written strings.

### 5. Multi-tenant SaaS: scoped tokens per customer

Issue each customer a token that can only see their own devices — no subaccount management needed for read-only fleet access:

```python
from flespi_client.core.enums import RESTEnum
from flespi_client.schemas.tokens.access import AclAccessSchema
from flespi_client.schemas.tokens.ace import GwDevicesAceSchema
from flespi_client.schemas.tokens.base import TokenSchema

customer_device_ids = [101, 102, 103]  # this customer's fleet only

token = (
    await client.platform.tokens().post(
        TokenSchema(
            info="customer-42, read-only fleet access",
            ttl=86400,  # rotates automatically; re-fetch before it expires
            access=AclAccessSchema(
                acl=[
                    GwDevicesAceSchema(
                        ids=customer_device_ids, methods=[RESTEnum.get]
                    )
                ]
            ),
        )
    )
)[0]

print(token.key)  # hand this to the customer's integration, not your master token
```

For full account isolation (billing, limits, its own sub-tokens) rather than just scoped read access, create a `client.platform.subaccounts()` per customer instead and issue tokens under it.

### 6. Real-time alerting via webhooks

Push a notification to your own backend the instant a device reports in, instead of polling:

```python
from flespi_client.schemas.webhooks.base import (
    WebhookFlespiPlatformConfigSchema,
    WebhookSchema,
    WebhookTriggerSchema,
)

device_id = 101

webhook = await client.platform.webhooks().post(
    WebhookSchema(
        name="new-message-alert",
        triggers=[
            WebhookTriggerSchema(topic=f"flespi/message/gw/devices/{device_id}")
        ],
        configuration=WebhookFlespiPlatformConfigSchema(
            uri="/gw/devices/all/messages",  # or any flespi-internal REST path
            body="%payload%",
        ),
    )
)
```

Use `WebhookCustomServerConfigSchema` instead to POST to your own external HTTPS endpoint rather than a flespi-internal one. Trigger topics follow flespi's own MQTT namespace (`flespi/message/...`, `flespi/state/.../telemetry/+`, ...) — explore available topics with the [MQTT Board](https://flespi.com/tools/mqtt-board) in the flespi panel.

## Architecture

Requests flow through four layers:

1. **Transport** (`flespi_client/core/http/transport.py`) — sends HTTP via `aiohttp`, returns the raw `{result, errors}` JSON envelope (`flespi_client/core/http/base.py:FlespiEnvelope`). Resource/scope code depends on `TransportProtocol` (`flespi_client/core/http/protocol.py`), not this concrete class — any object with the same shape (a test fake, your own implementation) works, no inheritance required.
2. **Error translation** (`flespi_client/core/http/decorators.py`) — maps flespi's `errors[]` envelope to typed `HTTPException` subclasses. A code this SDK doesn't recognize raises `UnknownFlespiError` rather than being silently reported as `InternalServerError`.
3. **Scope/resource graph** (`flespi_client/core/scopes/`) — fluent, chainable resource paths mirroring flespi's REST tree, e.g. `client.gw.calcs(PathSelector(FlespiField("id", Expr.EQ, 1))).devices(PathSelector())`. Path nesting and selector syntax are validated before any network call.
4. **`FlespiClient`** (`flespi_client/core/client.py`) — wires the layers together, exposes `.gw`, `.ai`, `.auth`, `.realm`, `.mqtt`, `.platform`, `.storage`.

Every resource call refreshes the token and retries once automatically on a 401 (`with_auto_refresh` in `flespi_client/core/scopes/base.py`, wired to `Transport.try_refresh` via `FlespiClient._try_refresh_token`) — this needs no opt-in and works for any `client.gw.foo().get()`/`post()`/`put()`/`patch()`/`delete()` call. It's a no-op (the original `UnauthorizedError` just propagates) for a token-only client with no `username`/`password`/`realm_id`. Concurrent calls hitting a stale token at once share one real refresh instead of each triggering their own (`Transport.try_refresh` is single-flight over an `asyncio.Task`). `FlespiClient.with_retry` remains for the few call paths that don't go through a resource verb method at all (e.g. code built on `raw_get()`).

Separately, `Transport` retries a real transient failure at the HTTP layer itself — a `429`/`500`/`502`/`503`/`504` status or a dropped connection — with capped-exponential, full-jitter backoff, or the server's own `Retry-After` header when it sends one (`flespi_client/core/http/retry.py:RetryPolicy`, backoff shape mirrors `LockManager`'s). This is unrelated to flespi's own `errors[]` business codes, which travel inside a normal `200` response and are never retried this way. Only `GET`/`PUT`/`PATCH`/`DELETE` are retried by default — `POST` creates a resource, so blindly resending one whose response merely got lost would risk creating a duplicate; opt a specific case in via `RetryPolicy(retry_methods=...)`. Tune or disable it via `FlespiClient(retry_policy=RetryPolicy(max_attempts=1))` (or pass one straight to `Transport`) — `retry_policy`/`observers` and an explicit `transport` are mutually exclusive constructor args (`FlespiClient` raises `ValueError` if both are given), since a caller-supplied transport's own setup governs instead.

Selectors (`{expression}` path segments) are built with a typed `Expr` AST (`flespi_client/core/selectors/expr.py`) via plain `PathSelector(expr)` — raw strings are not accepted, and an unknown field name is rejected before any network call for any resource with a known response schema. A comparison operator can be written either as the `Expr.EQ`-style constant or its raw symbol (`FlespiField("id", "==", 5)`) — both are validated against the same fixed set and normalized to the same value.

A destructive PUT/PATCH/DELETE that only makes sense against specific item(s) — essentially every write across `gw`/`platform`/`storage`/`mqtt` — requires a *non-empty* selector before any network call (`flespi_client/core/utils/resolutions.py:require_selector`, `flespi_client/core/scopes/base.py:check_method`). A bare `PathSelector()` (`{}`) doesn't satisfy this: flespi treats an empty expression the same as "no filter" (this SDK's own "list everything" idiom, e.g. `client.gw.devices(PathSelector()).get()`) — too easy to reach for by accident to also silently authorize `client.gw.devices(PathSelector()).delete()` against every device. The check cascades through the whole chain, not just the immediate call: `client.gw.groups().assets(PathSelector(id==5)).delete()` is rejected too, even though the `assets(...)` selector itself looks specific, because the bare `groups()` above it isn't. A selector-less passthrough hop (`Cid`, `Billing`, `.logs()`, ...) doesn't reset that chain — it just carries the ancestor's sufficiency through unchanged.

Separately, GET always requires *some* selector — even the empty `PathSelector()` counts, unlike the write check above. flespi registers no bare-collection GET route at all (only `/gw/devices/{dev-selector}`, `dev-selector` marked `required` in its own swagger) — `client.gw.devices()` (truly bare, no selector object at all) doesn't correspond to any real endpoint and is rejected the same way. `client.gw.devices().post(...)` stays exactly this bare, though: flespi's create endpoint is the separate route `POST /gw/devices`, with no selector segment at all — passing one would send to a URL that doesn't exist for POST. One `BaseResource` instance's path is fixed at construction and reused for whichever verb is called on it, so construct bare for a create, with a selector (even an empty one) for anything reading or targeting existing items.

Pydantic schemas validated against flespi's real API constraints (field lengths, numeric ranges, enums) live under `flespi_client/schemas/`, one package per resource (`flespi_client/schemas/channels/`, `flespi_client/schemas/devices/`, `flespi_client/schemas/tokens/`, ...), each split into `base.py` (POST body), `update.py` (PUT/PATCH body), and `response.py` (parsed GET/POST/PUT/PATCH result).

Virtually every resource — `client.gw.channels()`, `.devices()`, `.streams()`, `.calcs()`, `.groups()`, `.assets()`, `.geofences()`, `.storage.containers()`, `.storage.cdns()`, `.platform.webhooks()`, `.platform.tokens()`, `.platform.realms()`, and so on — validates its POST/PUT/PATCH body against the matching schema *and* parses every `get`/`post`/`put`/`patch` result into the matching response schema, automatically (`flespi_client/core/scopes/base.py:ValidatedResponseResource`). Pass a plain `dict`, the schema instance, or (for POST) a list of either; invalid bodies raise `pydantic.ValidationError` before any network call, and a schema instance built for the *wrong* resource (a `DeviceSchema` passed to `channels().post(...)`) is rejected by mypy too. A handful of action-style endpoints with no list-collection response (`ai/tools/*`, `realm/*/login`) use a related mechanism (`ValidatedActionResource`) that validates the same way but preserves the response's other top-level fields (e.g. `credits`) instead of unwrapping just `result`.

`fields`/`limit`/`offset` query params on `get()` (and the `fields`-only equivalent on `post()`/`patch()`, per flespi's own swagger — no `limit`/`offset` there) are validated the same way, against `flespi_client/schemas/base.py:ListQuerySchema`/`FieldsQuerySchema` — pass a plain `dict` or a schema instance.

Every opaque `dict[str, Any]` field — `metadata` on all schemas (`flespi_client/schemas/base.py:MetadataSchema`), plus `configuration`/`settings`/`telemetry`/`protocol_features` and the device-command fields (`fields`, `properties`, `tags`, `schema`, `meta`) — is an independent generic type parameter, defaulting to `dict[str, Any]`. Parametrize any subset to validate it against your own model instead of accepting an arbitrary JSON object; unparametrized ones keep defaulting to a plain dict. The *resource* itself is re-specialized the same way at the call site — `client.gw.channels[Metadata, Configuration]` — so the post/update/response schemas all pick up your types together:

```python
class MyMetadata(BaseModel):
    owner: str


class MyConfig(BaseModel):
    ident: str


# Channels[Metadata, Configuration] — parametrize only what you need
created = await client.gw.channels[MyMetadata, MyConfig]().post(
    ChannelSchema[MyMetadata, MyConfig](
        name="ch1",
        protocol_id=1,
        metadata=MyMetadata(owner="bob"),
        configuration=MyConfig(ident="123456"),
    )
)
channel = created[0]
print(channel.metadata.owner if channel.metadata else None)  # typed MyMetadata, not a dict
```

Write operations optionally take a distributed Redis lock (`flespi_client/core/locks/manager.py`) to avoid concurrent-write races across processes — held for the whole write, including any retry-with-backoff underneath it, so a flaky flespi can now stretch a write's critical section well past one HTTP round trip; pass a non-zero `retry_timeout_ms` (`LockManager` docstring) if a fail-fast lock under that contention matters for your write pattern.

Bulk-create `POST` bodies larger than flespi's per-endpoint item limit can be split automatically:

```python
await client.gw.geofences().post(many_geofences, chunk_size=128)
```

Each chunk is sent as its own request; results are concatenated in order. A chunk failing after earlier ones already succeeded raises `ChunkedPostError` (`flespi_client/core/http/errors.py`) instead of the plain underlying error — `.partial_results` holds what those earlier chunks already created, `.failed_chunk_index` which chunk broke, and `__cause__` why, so a caller can reconcile instead of re-querying everything. Every resource with a documented per-request item cap (`MAX_DEVICES_PER_REQUEST`, `MAX_GEOFENCES_PER_REQUEST`, ...) rejects a single oversized call with a clear `ValueError` up front instead of letting it reach the network only to be rejected there.

## Metrics

Every HTTP request made via `Transport` is instrumented through a pluggable `TransportObserver` (`flespi_client/core/http/observers.py`) — `Transport` fans each request/error/retry event out to every registered observer without knowing about any of them by name, isolating each one in its own try/except so a broken observer (logs and is skipped) never aborts the real request or blocks its siblings. The default, `PrometheusTransportObserver`, records with [prometheus-client](https://github.com/prometheus/client_python) (`flespi_client/core/metrics.py`):

- `flespi_sdk_requests_total{method,path,status}` — request count
- `flespi_sdk_request_duration_seconds{method,path}` — latency histogram
- `flespi_sdk_errors_total{method,path,flespi_code}` — business errors from flespi's `errors[]` envelope
- `flespi_sdk_retries_total{method,path,reason}` — retried requests (transient status or connection error)

`path` is normalized (`{expression}` selectors collapsed to `{selector}`) to keep label cardinality bounded. Metrics register on the default `prometheus_client` registry — expose them however your app already serves metrics (e.g. `prometheus_client.make_asgi_app()` / `start_http_server()`).

Add your own backend (OpenTelemetry, structured logging, ...) by passing additional observers — `FlespiClient(observers=[PrometheusTransportObserver(), MyOtelObserver()])` — or drop prometheus entirely by passing just your own.

## Development

```bash
uv sync
uv run pytest
uv run mypy .
uvx ruff check .
```
