flowion

The Python client for the Flowion API.

Beta. The API schema is not yet committed to and may change in any 0.MINOR release. Pin an exact version (flowion==0.1.0) and read the release notes before you move.

Async-first, typed (py.typed, checked with mypy --strict), and it needs no HTTP library of its own: you hand it a transport. Most of the package is generated from the API's OpenAPI contract. Supported Python: 3.12 and newer (verified on 3.12 and 3.14). Licensed MIT OR Apache-2.0.

Install

pip install flowion               # the SDK alone: standard library only
pip install "flowion[aiohttp]"    # plus the adapter for an aiohttp session

The transport

The SDK never opens a connection. Everything goes through one small protocol (flowion.Transport), so the HTTP stack, its pool and its TLS settings stay yours:

class Transport(Protocol):
    async def send(self, request: Request) -> Response: ...
    def stream(self, request: Request) -> AbstractAsyncContextManager[StreamResponse]: ...

send returns for every HTTP status (4xx and 5xx included) and raises only TransportError when no response arrived. It does not follow redirects, header names in Response.headers are lowercase, and asyncio.CancelledError must pass through untouched. Request carries method, url, headers, body and timeout; Response is status, headers, body; StreamResponse is status, headers and an async iterator of byte chunks.

A custom transport is a dozen lines. This is an illustration over httpx, not something the SDK ships or tests:

class HttpxTransport:
    def __init__(self, http: httpx.AsyncClient) -> None:
        self._http = http

    async def send(self, request: Request) -> Response:
        try:
            r = await self._http.request(request.method, request.url, headers=dict(request.headers), content=request.body, timeout=request.timeout)
        except httpx.HTTPError as error:
            raise TransportError(str(error)) from error
        return Response(r.status_code, {k.lower(): v for k, v in r.headers.items()}, r.content)

    # stream(): an @asynccontextmanager that yields a StreamResponse built the same way.

For aiohttp there is an adapter, and it works over a session you already own. Home Assistant passes its own; the adapter only calls session.request(...) with that one request's headers and timeout. It never closes the session, never changes its headers or cookie jar, and never touches its connector:

from flowion.aiohttp_transport import AiohttpTransport

session = async_get_clientsession(hass)   # Home Assistant's shared session
transport = AiohttpTransport(session)

The Flowion gateway is bearer-only and sets no cookies, which is what makes sharing a session safe. The adapter cannot stop aiohttp sending a cookie the session already holds for that host.

First call

Every operation is an async method on a resource group: client.<group>.<method>(...). Path parameters come first, in path order; everything else is a keyword. token is an API key or access token as a string:

from flowion import Client

client = Client(transport, token="flw_ak_your_key_here")
project_id = "11111111-1111-4111-8111-111111111111"

page = await client.projects.charging_stations.list_chargers(project_id, limit=25)
for charger in page.data:
    print(charger.charge_point_id, charger.status)

(transport is any Transport, for example AiohttpTransport(session).) Python keywords and builtins get a trailing underscore in field names: the id of a charger is charger.id_. Client(..., mode="sandbox") sends X-Flowion-Mode; timeout_seconds= sets the per-request timeout (30 s by default).

A token that expires should not be a string. Pass a TokenProvider, an async callable awaited once per request, so your own refresh logic runs under your own lock:

async def current_token() -> str:
    return "the-current-access-token"   # read it from your token store, refreshing if needed

client = Client(transport, token=current_token)
page = await client.projects.charging_stations.list_chargers(project_id)

A 401 with code invalidToken raises ApiError. The client does not refresh and does not retry it; that is the caller's decision.

The grant for a public client

flowion.oauth.OAuthClient implements the authorization-code grant with PKCE for a public client: no client secret, ever. Three steps, and the SDK stores nothing between them: you keep state and code_verifier yourself.

For Home Assistant, the redirect URI is https://my.home-assistant.io/redirect/oauth and the scopes are chargers:read prices:read sessions:read. The client_id is the one registered for your integration with Flowion.

from flowion.oauth import OAuthClient

oauth = OAuthClient(
    transport,
    client_id="your-client-id",
    redirect_uri="https://my.home-assistant.io/redirect/oauth",
)
request = oauth.authorization_request(scopes=["chargers:read", "prices:read", "sessions:read"])
print(request.url)   # send the user's browser here; keep request.state and request.code_verifier

The user approves on Flowion and the browser comes back to the redirect URI. parse_redirect accepts the full URL or just its query, checks state in constant time and sends nothing. Then exchange the code, call the API, and refresh before the access token expires:

from flowion import Client

# redirect_url: the full URL the user's browser came back to
result = oauth.parse_redirect(redirect_url, expected_state=request.state)
tokens = await oauth.exchange_code(result.code, code_verifier=request.code_verifier)

client = Client(transport, token=tokens.access_token)
async for project in client.connection.connection_projects_all():
    print(project.id_, project.name)

# tokens.expires_at is an aware datetime; refresh ahead of it.
tokens = await oauth.refresh(tokens.refresh_token)   # the old refresh token is now spent

Things to know:

• Refresh is rotating. Keep the new refresh_token from every refresh. Replaying a spent one ends the whole grant (InvalidGrant: authorize again). Two concurrent refresh calls on one token race into exactly that, so serialise refresh behind one lock, as Home Assistant's OAuth2Session does.
• Errors map to types: AccessDenied (the user declined), StateMismatch, InvalidGrant, InvalidClient, InvalidRequest, all subclasses of OAuthError with the RFC 6749 code.
• The Home Assistant data routes are three read-only calls under /v1/connection/projects/{projectId}/: chargers, price and sessions. They land with the platform release that adds them (spec 166) and are not in this SDK release, so it has no method for them yet; they appear in the first SDK regenerated after that release ships. Until then the grant above works, and client.connection offers the calls the contract has today.

Errors

Everything the SDK raises on purpose derives from flowion.FlowionError:

• TransportError: no response arrived (connect, read, TLS, timeout). Never retried.
• RateLimitedError: a 429 that was not, or no longer could be, retried. retry_after_seconds and rate_limit (limit, remaining, reset_seconds).
• ApiError: any other non-2xx answer with the platform's error body. status, code (the platform's string, verbatim), body and request_id.
• DecodeError: a body the SDK cannot decode (a proxy's HTML page, truncated JSON); carries status, request_id and a short excerpt.
• OAuthError and its subclasses, above.

try:
    await client.projects.charging_stations.list_chargers(project_id)
except RateLimitedError as error:
    print("slow down for", error.retry_after_seconds, "s")
except ApiError as error:
    print(error.status, error.code, "request id", error.request_id)   # quote it when you report a problem

Retry is on by default and covers 429 only: up to 4 attempts in total, waiting the platform's Retry-After (1 s if it sent none). A Retry-After above the 30 s cap is not waited for: the error surfaces at once so you decide. A 5xx or a transport error is never retried, because a write may already have been processed. Tune it with RetryPolicy(max_attempts=..., max_wait_seconds=...); max_attempts=1 turns it off. client.last_rate_limit holds the rate-limit headers of the latest response.

Values the SDK has not seen

The API can grow a new enum value or a new variant of a union without the SDK changing. Such a response is kept, not rejected: an open enum is typed Literal[...] | str, and an open union gains an ...Unrecognized form holding the raw data. For each there is a guard that narrows to the known values:

from flowion.generated.types import is_charger_status_known

for charger in page.data:
    if is_charger_status_known(charger.status):
        ...   # the type checker now knows it is "online" or "offline"
    else:
        ...   # a status added after this release; charger.status is the raw string

Pagination and live events

Each list operation has an *_all twin that returns an async iterator over every item, fetching the next page only when you read past the last one:

async for charger in client.projects.charging_stations.list_chargers_all(project_id, status="online"):
    print(charger.charge_point_id)

A live-event stream is an async iterator too. It ends on a clean close or when the server revokes the stream, and an event type the SDK does not know is yielded as the unrecognized form instead of raising into your loop:

async for event in client.projects.live_events.stream_live_events(project_id):
    print(event.charging_station_id, event.type_)

What is not provided

• The confidential-client path (HTTP Basic client secret). This SDK is for public clients.
• The browser consent routes (/v1/connections/authorize/*), which take a signed-in session, not a bearer token. The SDK builds the authorization URL; the user opens it.
• Token storage, caching or persistence, and opening a browser. You keep the state, the verifier and the tokens.
• A synchronous facade. Run the client on an event loop.
• A default HTTP stack. Bring a Transport, or install the aiohttp extra.

Security

Tokens, API keys, authorization codes and PKCE verifiers are never put in logs or in configuration in clear, and the SDK never logs them: not in messages, not in repr, not in exceptions. The flowion logger emits DEBUG lines with method, path, status and request id only. Keep your own tokens out of your logs and store them in your platform's secret storage.

What has been verified

The package is exercised against a mock transport, against the shared shared conformance scenarios that every Flowion SDK plays, and, for the grant, against a local server that plays the platform's side. Runs against production are checked by the maintainers for each release, not by the test suite.

Licence

MIT OR Apache-2.0, at your option. See LICENSE-MIT and LICENSE-APACHE.