pybragerone.api.client

HTTP API client for BragerOne.

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.

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]

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_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_connect(wsid_ns: str, modules: list[str], group_id: int | None = None, engine_sid: str | None = None)[source]

Connect to modules via WebSocket.

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