pybragerone.api.client

HTTP API client for BragerOne.

Functions

format_expected_failure_reason(err)

Return a compact log reason: exception type plus HTTP status when known.

is_expected_upstream_unavailable(err)

Return whether err is an expected upstream capacity/maintenance failure.

Classes

BragerOneApiClient(*[, server, api_base, ...])

HTTP API client with idempotent login and auto-refresh.

HttpCache()

Simple HTTP cache using ETag/Last-Modified headers with in-memory body storage.

Exceptions

ApiError(status, data[, headers])

Raised for non-2xx HTTP responses.
exception pybragerone.api.client.ApiError(status: int, data: Any, headers: dict[str, str] | None = None)[source]

Bases: RuntimeError

Raised for non-2xx HTTP responses.

Variables:

  • status – HTTP status code.

  • data – Response data.

  • headers – Response headers.

Parameters:
class pybragerone.api.client.BragerOneApiClient(*, server: ServerConfig | None = None, api_base: str = 'https://io.brager.pl/v1', one_base: str = 'https://one.brager.pl', io_base: str | None = None, token_store: TokenStore | None = None, enable_http_trace: bool = False, redact_secrets: bool = True, creds_provider: Callable[[], tuple[str, str]] | None = None, validate_on_start: bool = True, refresh_leeway: int = 90, timeout: float = 8.0, concurrency: int = 4, verify: bool | str | SSLContext = True)[source]

Bases: object

HTTP API client with idempotent login and auto-refresh.

Provides a comprehensive HTTP client for the BragerOne API with automatic token management, authentication, and retry logic.

Parameters:
property access_token: str

Get the current access token.

Returns:

Current access token string.

Raises:

RuntimeError – If the client is not authenticated yet.

property api_base: str

Base URL for REST API calls.

async close()[source]

Close the underlying HTTP session.

This should be called when the client is no longer needed to properly release resources.

Return type:

None

async ensure_auth(email: str | None = None, password: str | None = None)[source]

Ensure valid token: use cache + validation, and if missing/expired — login.

Parameters:

  • email (str | None) – User email for authentication.

  • password (str | None) – User password for authentication.

Returns:

Token – Valid authentication token.

Raises:

ApiError – If authentication fails or credentials are missing.

Return type:

Token

async fetch_json_one(path: str)[source]

Fetch JSON content from ONE_BASE.

Parameters:

path (str) – The path to fetch from.

Returns:

tuple[int, dict[str, Any] | list[Any] | None] – Tuple of (status_code, json_data).

Return type:

tuple[int, dict[str, Any] | list[Any] | None]

Note

Assets are JS, so JSON extraction should be done catalog-side, not here. This method is here only for symmetry; you may not use it.

async fetch_text_one(path: str)[source]

Fetch text content from ONE_BASE.

Parameters:

path (str) – The path to fetch from.

Returns:

tuple[int, str] – Tuple of (status_code, text_content).

Return type:

tuple[int, str]

async get_bytes(url: str)[source]

Get raw bytes from URL with caching (ETag/Last-Modified).

Parameters:

url (str) – The URL to fetch.

Returns:

bytes – The raw bytes of the response.

Raises:

ApiError – If the request fails.

Return type:

bytes

Note

This method uses an in-memory cache to store ETag and Last-Modified headers to optimize subsequent requests to the same URL.

async get_module_card(code: str)[source]

Get module card information by code.

Parameters:

code (str) – The module code identifier.

Returns:

ModuleCard – Module card information as a Pydantic model.

Raises:

ApiError – If the request fails.

Return type:

ModuleCard

async get_modules(object_id: int)[source]

GET /v1/modules?page=1&limit=999&group_id=<object_id> → list of modules.

Parameters:

object_id (int) – The object/group identifier.

Returns:

list[Module] – List of Module models (invalid rows are skipped and logged).

Raises:

ApiError – If the HTTP status is not 200 (empty list used to be ambiguous with a genuine empty object listing).

Return type:

list[Module]

async get_object(object_id: int)[source]

Get object by ID.

Parameters:

object_id (int) – The object identifier.

Returns:

ObjectDetails – Object details with operational status.

Raises:

ApiError – If the request fails.

Return type:

ObjectDetails

async get_object_permissions(object_id: int)[source]

Get object-specific permissions for the current user.

Parameters:

object_id (int) – The ID of the object to get permissions for.

Returns:

list[Permission] – List of object-specific permissions.

Raises:

ApiError – If the request fails.

Return type:

list[Permission]

async get_objects()[source]

GET /v1/objects → list of objects (with tolerance for different shapes).

Returns:

list[BragerObject] – List of BragerObject models.

Return type:

list[BragerObject]

async get_system_version()[source]

Get system version information.

Returns:

SystemVersion – System version information as a Pydantic model.

Raises:

ApiError – If the request fails.

Return type:

SystemVersion

async get_user()[source]

Get current user information.

Returns:

User – User information as a Pydantic model.

Raises:

ApiError – If the request fails.

Return type:

User

async get_user_permissions()[source]

Get current user permissions.

Returns:

list[Permission] – List of user permissions.

Raises:

ApiError – If the request fails.

Return type:

list[Permission]

async invalidate_and_reauth(email: str | None = None, password: str | None = None)[source]

Drop the cached token and force a fresh login.

Used by zombie-session recovery when reconnecting with the existing (still-valid) access token fails to restore live WS ParamUpdate traffic. Credentials are resolved before any destructive clear: if neither explicit args nor creds_provider can supply both email and password, a usable in-memory token is left intact (soft path — log a warning and return it) so reconnect is not left unauthenticated. Only when credentials are available do invalidation, optional persisted-store clear, and _post_login run under one _auth_lock hold so a racing ensure_auth cannot interleave a provider login ahead of explicit overrides. The clearer runs via asyncio.to_thread (keyring/fs I/O).

Parameters:

  • email (str | None) – Optional email override for the login call.

  • password (str | None) – Optional password override for the login call.

Returns:

Token – Fresh authentication token from login, or the existing token when credentials are unavailable and a usable token is already cached.

Raises:

ApiError – If credentials are missing and there is no usable token.

Return type:

Token

property io_base: str

Base URL for Engine.IO/Socket.IO server.

async module_command(*, devid: str, pool: str, parameter: str, value: Any, parameter_name: str | None = None, unit: int | None = None, extra_payload: Mapping[str, Any] | None = None, return_data: bool = False)[source]

Dispatch a parameter-like command payload to a module.

This method posts to /v1/module/command and mirrors the web application’s payload shape for non-raw writes.

Parameters:

  • devid (str) – Module device identifier.

  • pool (str) – Parameter pool/group (for example "P4").

  • parameter (str) – Parameter channel/index token (for example "v8").

  • value (Any) – Value to be written.

  • parameter_name (str | None) – Optional display/name token.

  • unit (int | None) – Optional unit identifier expected by backend for selected writes.

  • extra_payload (Mapping[str, Any] | None) – Optional additional keys to include in the request payload.

  • return_data (bool) – Whether to return (status, data) instead of only success.

Returns:

tuple[int, Any] | bool – Tuple of (status, data) when return_data=True; otherwise a boolean indicating request success.

Return type:

tuple[int, Any] | bool

async module_command_auto(*, devid: str, value: Any | None = None, command: str | None = None, pool: str | None = None, parameter: str | None = None, parameter_name: str | None = None, unit: int | None = None, extra_payload: Mapping[str, Any] | None = None, return_data: bool = False)[source]

Automatically dispatch via /module/command or /module/command/raw.

Routing rules:

  • If command is provided, dispatches through /module/command/raw.

  • If both pool and parameter are provided, dispatches through /module/command.

  • If both route styles are provided at once, raises ValueError.

  • If no complete route style is provided, raises ValueError.

Parameters:

  • devid (str) – Module device identifier.

  • value (Any | None) – Command/write value. Optional for raw commands.

  • command (str | None) – Symbolic command for /module/command/raw.

  • pool (str | None) – Parameter pool/group for /module/command.

  • parameter (str | None) – Parameter channel/index token for /module/command.

  • parameter_name (str | None) – Optional display/name token for /module/command.

  • unit (int | None) – Optional unit identifier for /module/command.

  • extra_payload (Mapping[str, Any] | None) – Optional additional keys to include in request payload.

  • return_data (bool) – Whether to return (status, data) instead of only success.

Returns:

tuple[int, Any] | bool – Tuple of (status, data) when return_data=True; otherwise a boolean indicating request success.

Return type:

tuple[int, Any] | bool

async module_command_raw(*, devid: str, command: str, value: Any | None = None, extra_payload: Mapping[str, Any] | None = None, return_data: bool = False)[source]

Dispatch a symbolic command to a module.

This method matches the web application’s command transport and posts to /v1/module/command/raw.

Parameters:

  • devid (str) – Module device identifier.

  • command (str) – Symbolic command name (for example "MODULE_RESTART").

  • value (Any | None) – Optional command value (for example "ONLINE" or "OFF").

  • extra_payload (Mapping[str, Any] | None) – Optional additional keys to include in the request payload.

  • return_data (bool) – Whether to return (status, data) instead of only success.

Returns:

tuple[int, Any] | bool – Tuple of (status, data) when return_data=True; otherwise a boolean indicating request success.

Return type:

tuple[int, Any] | bool

async modules_activity(modules: list[str], *, page: int = 1, limit: int = 20, return_data: bool = False)[source]

Fetch a page of module activity rows (SPA activity list).

Parameters:

  • modules (list[str]) – Module device identifiers.

  • page (int) – 1-based page index (SPA default).

  • limit (int) – Page size (SPA default 20).

  • return_data (bool) – When true, return (status, data).

Returns:

tuple[int, Any] | bool – Success flag, or (status, payload) when return_data is set.

Return type:

tuple[int, Any] | bool

async modules_activity_quantity_prime(modules: list[str], *, return_data: bool = False)[source]

Prime modules activity quantity.

Parameters:

  • modules (list[str]) – List of module codes.

  • return_data (bool) – Whether to return response data or just success status.

Returns:

tuple[int, Any] | bool – Tuple of (status, data) if return_data=True, otherwise boolean success.

Return type:

tuple[int, Any] | bool

async modules_alarms(modules: list[str], *, page: int = 1, limit: int = 20, return_data: bool = False)[source]

Fetch active module alarms (SPA current-alarms list).

Parameters:

  • modules (list[str]) – Module device identifiers.

  • page (int) – 1-based page index.

  • limit (int) – Page size.

  • return_data (bool) – When true, return (status, data).

Returns:

tuple[int, Any] | bool – Success flag, or (status, payload) when return_data is set.

Return type:

tuple[int, Any] | bool

async modules_alarms_history(modules: list[str], *, page: int = 1, limit: int = 20, return_data: bool = False)[source]

Fetch finished/historical module alarms (SPA alarm-history list).

Parameters:

  • modules (list[str]) – Module device identifiers.

  • page (int) – 1-based page index.

  • limit (int) – Page size.

  • return_data (bool) – When true, return (status, data).

Returns:

tuple[int, Any] | bool – Success flag, or (status, payload) when return_data is set.

Return type:

tuple[int, Any] | bool

async modules_alarms_quantity(modules: list[str], *, return_data: bool = False)[source]

Fetch per-module active alarm badge counts.

Parameters:

  • modules (list[str]) – Module device identifiers.

  • return_data (bool) – When true, return (status, data).

Returns:

tuple[int, Any] | bool – Success flag, or (status, payload) when return_data is set.

Return type:

tuple[int, Any] | bool

async modules_connect(wsid_ns: str, modules: list[str], group_id: int | None = None, engine_sid: str | None = None)[source]

Connect to modules via WebSocket.

Remembers which JSON shape succeeded (wsid vs sid, optional group_id) but always injects the current namespace/engine SID. Replaying a cached body with a stale SID after reconnect was observed to return HTTP 200 while leaving the new Socket.IO session unbound.

Parameters:

  • wsid_ns (str) – WebSocket namespace ID.

  • modules (list[str]) – List of module codes to connect to.

  • group_id (int | None) – Optional group/object ID.

  • engine_sid (str | None) – Optional engine session ID as fallback.

Returns:

bool – True if connection successful, False otherwise.

Return type:

bool

async modules_parameters_prime(modules: list[str], *, return_data: bool = False)[source]

Prime modules parameters.

Parameters:

  • modules (list[str]) – List of module codes.

  • return_data (bool) – Whether to return response data or just success status.

Returns:

tuple[int, Any] | bool – Tuple of (status, data) if return_data=True, otherwise boolean success.

Return type:

tuple[int, Any] | bool

property one_base: str

Base URL for web app/assets origin.

async revoke()[source]

Server-side logout + local cleanup of token and persistence.

Attempts to revoke the token on the server and cleans up local state regardless of server response.

Return type:

None

set_token_store(store: TokenStore | None)[source]

Wire token persistence to a store that exposes load/save/clear.

Parameters:

store (TokenStore | None) – Token storage instance or None to disable persistence.

Return type:

None

class pybragerone.api.client.HttpCache[source]

Bases: object

Simple HTTP cache using ETag/Last-Modified headers with in-memory body storage.

This cache implementation stores response bodies along with their cache headers to enable efficient HTTP caching with proper conditional request handling.

get_body(url: str)[source]

Retrieve cached response body for URL.

Parameters:

url (str) – The URL to retrieve the cached body for.

Returns:

bytes | None – Cached response body as bytes, or None if not cached.

Return type:

bytes | None

headers_for(url: str)[source]

Generate conditional request headers for cached URL.

Returns appropriate If-None-Match and If-Modified-Since headers based on cached ETag and Last-Modified values for the given URL.

Parameters:

url (str) – The URL to generate headers for.

Returns:

dict[str, str] – Dictionary of conditional headers to include in the request.

Return type:

dict[str, str]

update(url: str, headers: Mapping[str, str], body: bytes)[source]

Update cache entry with response data.

Stores the response body along with ETag and Last-Modified headers for the given URL to enable future conditional requests.

Parameters:

  • url (str) – The URL being cached.

  • headers (Mapping[str, str]) – Response headers containing cache information.

  • body (bytes) – The response body to cache.

Return type:

None

pybragerone.api.client.format_expected_failure_reason(err: BaseException)[source]

Return a compact log reason: exception type plus HTTP status when known.

Avoids ApiError’s default data!r message, which often embeds large upstream HTML bodies and would spam logs during extended outages.

Return type:

str

Parameters:

err (BaseException)

pybragerone.api.client.is_expected_upstream_unavailable(err: BaseException)[source]

Return whether err is an expected upstream capacity/maintenance failure.

Matches ApiError / aiohttp response errors with HTTP 502/503/504, including when wrapped in ExceptionGroup or __cause__ / __context__ chains.

Return type:

bool

Parameters:

err (BaseException)