Clients¶
BusyBar is the synchronous client and AsyncBusyBar is its async/await
counterpart. Both compose the same set of endpoint mixins, so the method
surface is identical apart from the coroutines.
from busylib import AsyncBusyBar, BusyBar
bb = BusyBar("10.0.4.20")
async_bb = AsyncBusyBar(addr="10.0.4.20", token="my-access-key")
Creating either client has no terminal output and does not contact the device. The first endpoint call performs the connection; see the quick start for its expected output.
BusyBar(addr: str | None = None, *, token: str | None = None, timeout: float | Timeout | None = None, max_retries: int = 2, backoff: float = DEFAULT_BACKOFF, transport: BaseTransport | None = None, api_version: str | None = None, compatibility_mode: CompatibilityMode = 'warn', is_cloud: bool | None = None)
Bases: AccessMixin, AccountMixin, BusyMixin, TimeMixin, UpdaterMixin, FirmwareMixin, StorageMixin, AssetsMixin, DisplayMixin, AudioMixin, WifiMixin, InputMixin, SmartHomeMixin, StateStreamMixin, BleMixin, SyncClientBase
HTTPX-based client for the BUSY Bar API.
Build a client for one bar.
addr is a device address; leaving it out with a token reaches the
bar through the cloud, at the host BUSYLIB_CLOUD_URL names. Pass
is_cloud=True only to name a cloud host per client, which is what
stops an address like api.dev.busy.app being taken for a device.
property
¶
is_usb_connected: bool
Returns True if a USB device was found and connected.
class-attribute
instance-attribute
¶
connection_type: Literal['local', 'cloud', 'network'] = 'network'
instance-attribute
¶
client = httpx2.Client(base_url=self.base_url, headers=headers or None, timeout=_as_timeout(timeout), transport=transport)
property
¶
is_cloud: bool
Check whether connection uses cloud mode.
Returns True for cloud connection_type.
property
¶
is_local: bool
Check whether connection uses local mode.
Returns True for local connection_type.
usb_reboot(*, raise_on_error: bool = False) -> bool
Attempt to reboot the device via USB.
Returns True on success and False on failure by default. If raise_on_error is True, re-raises BusyBarUsbError.
usb_reset(*, raise_on_error: bool = False) -> bool
Alias for usb_reboot().
Provided for callers that prefer "reset" naming.
method_compatibility(method_name: str) -> versioning.MethodCompatibility | None
Return declarative OpenAPI compatibility metadata for a client method.
is_local_available() -> bool
Check local API reachability on base_url.
Returns True when /api/version responds without network errors.
api_request(method: str, path: str, *, params: dict[str, Any] | None = None, headers: dict[str, str] | None = None, session_id: str | None = None, application_name: str | None = None, json_payload: JsonType | None = None, data: bytes | Iterable[bytes] | None = None, expect_bytes: bool = False, allow_text: bool = False, timeout: float | Timeout | None = None) -> JsonType | bytes | str
Execute a raw API request through the current client session.
Advanced callers can control path, params, headers, body, and request context without creating a separate HTTP client.
prepare_request(method: str, path: str, *, params: dict[str, Any] | None = None, headers: dict[str, str] | None = None, session_id: str | None = None, application_name: str | None = None, json_payload: JsonType | None = None, data: bytes | Iterable[bytes] | None = None, expect_bytes: bool = False, allow_text: bool = False, timeout: float | Timeout | None = None) -> PreparedRequest
Build a prepared request without executing network I/O.
External integrations can inspect the prepared payload, route it to
custom transports, or execute later via execute_prepared_request.
execute_prepared_request(prepared: PreparedRequest, *, client: Client | None = None) -> JsonType | bytes | str
Execute a previously prepared request.
By default the current httpx2.Client is used. Callers may inject a
custom client while preserving error mapping. Prepared streaming content
is single-use and should be regenerated for repeated executions.
stream_status_ws() -> None
Stream device status updates via WebSocket /api/status/ws.
Sync iteration is not supported because websocket lifecycle and frame handling are asynchronous in this client implementation.
smart_home_pairing() -> types.SmartHomePairingInfo
Fetch smart home pairing status via GET /api/smart_home/pairing.
smart_home_pairing_start() -> types.SmartHomePairingPayload
Start smart home pairing via POST /api/smart_home/pairing.
smart_home_pairing_stop() -> types.SuccessResponse
Stop smart home pairing via DELETE /api/smart_home/pairing.
smart_home_switch() -> types.SmartHomeSwitchState
Fetch smart home switch state via GET /api/smart_home/switch.
smart_home_switch_set(state: bool, *, startup: Literal['off', 'on', 'toggle', 'last'] | None = None) -> types.SuccessResponse
Set smart home switch state via POST /api/smart_home/switch.
wifi_enable() -> types.SuccessResponse
Removed from the device API.
No supported firmware serves POST /api/wifi/enable, so calling this raises
BusyBarRemovedEndpointError. Use wifi_connect() / wifi_disconnect() instead.
wifi_disable() -> types.SuccessResponse
Removed from the device API.
No supported firmware serves POST /api/wifi/disable, so calling this raises
BusyBarRemovedEndpointError. Use wifi_connect() / wifi_disconnect() instead.
wifi_networks() -> types.NetworkResponse
Scan for nearby networks via GET /api/wifi/networks.
The device cannot scan while it is associated: doing so returns
400 "Scan not possible when connected" as a BusyBarAPIError.
Disconnect first with wifi_disconnect(), or skip the scan and pass
the SSID to wifi_connect() directly.
audio_play(*, path: str | None = None, stock_path: str | None = None, payload: AudioPlayRequest | dict[str, Any] | None = None, **request_kwargs: Unpack[RequestKwargs]) -> types.SuccessResponse
Play audio through POST /api/audio/play.
The endpoint payload may reference either an uploaded asset path or a
stock path. Explicit path and stock_path keyword arguments override
the same keys provided in payload. Request context such as
application_name and session_id is accepted through request kwargs;
use api_request for fully custom bodies.
audio_stop() -> types.SuccessResponse
Stop audio through DELETE /api/audio/play.
Uses API-like naming for callers that mirror firmware endpoints.
display_draw(display_data: DisplayElements | dict[str, Any], *, clear_before_draw: bool = False, sanitize_text: bool = False, **request_kwargs: Unpack[RequestKwargs]) -> types.SuccessResponse
display(display_data: DisplayElements | dict[str, Any], *, clear_before_draw: bool = False, sanitize_text: bool = False, audio_payload: AudioPlayRequest | dict[str, Any] | None = None, **request_kwargs: Unpack[RequestKwargs]) -> types.SuccessResponse
Render display content and optionally play audio after draw.
Operations are sequential, not atomic: clear may succeed before draw fails, and audio failure may occur after display content is visible. Exceptions include the failed endpoint path for diagnostics.
display_clear(**request_kwargs: Unpack[RequestKwargs]) -> types.SuccessResponse
Clear display content through DELETE /api/display/draw.
Uses API-like naming for callers that mirror firmware endpoints.
screen(display_id: DisplaySpecLike) -> bytes
Fetch a single display frame via GET /api/screen.
Returns RGB bytes, three per pixel: 3456 for the front display (72x16) and 38400 for the back (160x80, sent L4-packed and expanded to grey triples here). The device sends colour as BGR and it is swapped on the way out.
The response body is base64-encoded, uncompressed framebuffer bytes
(the Content-Type: image/bmp header is misleading, there is no
real BMP header).
frame(display_id: DisplaySpecLike) -> Frame
Fetch a display frame as a Frame via GET /api/screen.
The same bytes as screen(), with the geometry and the display
attached, so callers can read pixels, rows or a PNG without tracking
the layout themselves.
assets_upload(application_name: str, filename: str, data: bytes, *, timeout: float | Timeout | None = ASSET_UPLOAD_TIMEOUT) -> types.SuccessResponse
Upload an asset file for the given application.
Uses a longer default timeout to tolerate large payload uploads.
storage_write(path: str, data: bytes, *, timeout: float | None = 60.0, progress_callback: Callable[[int, int], None] | None = None, chunk_size: int = 64 * 1024) -> types.SuccessResponse
storage_mkdir(path: str) -> types.SuccessResponse
Create a storage directory via POST /api/storage/mkdir.
storage_rename(old_path: str, new_path: str) -> types.SuccessResponse
Rename a storage entry via POST /api/storage/rename.
transport() -> types.NetworkInterfaceInfo
Fetch active network transport via GET /api/transport.
status_device() -> types.StatusDevice
Fetch device manufacturing status via GET /api/status/device.
status_firmware() -> types.StatusFirmware
Fetch firmware status via GET /api/status/firmware.
status_system() -> types.StatusSystem
Fetch runtime status via GET /api/status/system.
log_dump(filename: str | None = None) -> types.LogDumpResponse
Dump the in-memory device log buffer to a storage file.
filename is a bare name without a path or extension, matching
^[a-zA-Z0-9_-]+$ on firmware OpenAPI 25.0.0+; the device appends its
own extension and storage path. When omitted, the device picks a
default file.
Breaking change: prior to 25.0.0 this method accepted path= (a full
device-side path). That parameter has been removed rather than
aliased, since the two contracts are not translatable (a full path
never matches the new filename pattern). Callers targeting firmware
older than 25.0.0 should pin an older busylib release instead of
adapting call sites.
update(firmware_data: bytes) -> types.SuccessResponse
Upload firmware update TAR and initiate update.
update_status() -> types.UpdateStatus
Get firmware update status with progress information.
update_changelog(version: str) -> types.UpdateChangelogResponse
Fetch update changelog for a specific version.
update_install(version: str) -> types.SuccessResponse
Start firmware update installation by version.
update_abort_download() -> types.SuccessResponse
Abort an ongoing firmware download.
update_autoupdate() -> types.AutoupdateSettings
Fetch autoupdate settings via GET /api/update/autoupdate.
update_autoupdate_set(settings: AutoupdateSettings | dict[str, object]) -> types.SuccessResponse
Set autoupdate settings via POST /api/update/autoupdate.
time_timezone_info() -> types.TimezoneInfo
Fetch current device timezone via GET /api/time/timezone.
time_timezone_list() -> types.TimezoneListResponse
Fetch supported device timezones via GET /api/time/tzlist.
time_timestamp(timestamp: str) -> types.SuccessResponse
Set device time via POST /api/time/timestamp.
time_timezone(timezone: str) -> types.SuccessResponse
Set device timezone via POST /api/time/timezone.
busy_snapshot() -> types.BusySnapshot
Fetch busy snapshot via GET /api/busy/snapshot.
busy_snapshot_set(snapshot: BusySnapshot) -> types.SuccessResponse
Set busy snapshot via PUT /api/busy/snapshot.
busy_profile(slot: BusyProfileSlot) -> types.BusyProfile
Fetch a busy profile slot via GET /api/busy/profiles/{slot}.
busy_profile_set(slot: BusyProfileSlot, profile: BusyProfile | dict[str, object]) -> types.SuccessResponse
Set a busy profile slot via PUT /api/busy/profiles/{slot}.
account_unlink() -> types.SuccessResponse
Unlink the device from the account via DELETE /api/account.
account_link() -> types.AccountLink
Request an account link code via POST /api/account/link.
account_info() -> types.AccountInfo
Fetch linked account info via GET /api/account/info.
account_status() -> types.AccountState
Fetch MQTT status info via GET /api/account/status.
account_backend() -> types.AccountBackend
Fetch MQTT backend settings via GET /api/account/backend.
account_backend_set(backend: AccountBackend | dict[str, object]) -> types.SuccessResponse
Set MQTT backend settings via PUT /api/account/backend.
account_profile() -> types.AccountProfile
Removed from the device API.
No supported firmware serves GET /api/account/profile, so calling this raises
BusyBarRemovedEndpointError. Use account_backend() instead.
account_profile_set(profile: AccountProfileName, custom_url: str | None = None) -> types.SuccessResponse
Removed from the device API.
No supported firmware serves POST /api/account/profile, so calling this raises
BusyBarRemovedEndpointError. Use account_backend_set() instead.
access_set(mode: HttpAccessMode, key: str) -> types.SuccessResponse
Set HTTP access mode via POST /api/access.
access_tokens_list() -> types.AccessTokensInfo
List issued access tokens via GET /api/access/tokens.
The secret itself is returned only when a token is minted, so every
token here has token set to None.
Experimental: needs firmware from busy-app/busybar-firmware#886, which is not released yet, so it answers 404 on every released firmware.
access_token_mint(name: str) -> types.AccessToken
Issue a new access token via POST /api/access/tokens.
This is the only time the device discloses the secret; it cannot be read back afterwards.
Experimental: needs firmware from busy-app/busybar-firmware#886, which is not released yet, so it answers 404 on every released firmware.
access_tokens_delete_all() -> types.SuccessResponse
Revoke every access token via DELETE /api/access/tokens.
Experimental: needs firmware from busy-app/busybar-firmware#886, which is not released yet, so it answers 404 on every released firmware.
access_tokens_revoke(short_id: str) -> types.SuccessResponse
Revoke one access token via DELETE /api/access/tokens/{short_id}.
Experimental: needs firmware from busy-app/busybar-firmware#886, which is not released yet, so it answers 404 on every released firmware.
AsyncBusyBar(addr: str | None = None, *, token: str | None = None, timeout: float | Timeout | None = None, max_retries: int = 2, backoff: float = DEFAULT_BACKOFF, transport: AsyncBaseTransport | None = None, api_version: str | None = None, compatibility_mode: CompatibilityMode = 'warn', is_cloud: bool | None = None)
Bases: AsyncAccessMixin, AsyncAccountMixin, AsyncBusyMixin, AsyncTimeMixin, AsyncUpdaterMixin, AsyncFirmwareMixin, AsyncStorageMixin, AsyncAssetsMixin, AsyncDisplayMixin, AsyncAudioMixin, AsyncWifiMixin, AsyncInputMixin, AsyncSmartHomeMixin, AsyncStateStreamMixin, AsyncBleMixin, AsyncClientBase
Async HTTPX-based client for the BUSY Bar API.
Build a client for one bar.
addr is a device address; leaving it out with a token reaches the
bar through the cloud, at the host BUSYLIB_CLOUD_URL names. Pass
is_cloud=True only to name a cloud host per client, which is what
stops an address like api.dev.busy.app being taken for a device.
property
¶
is_usb_connected: bool
Returns True if a USB device was found and connected.
class-attribute
instance-attribute
¶
connection_type: Literal['local', 'cloud', 'network'] = 'network'
instance-attribute
¶
client = httpx2.AsyncClient(base_url=self.base_url, headers=headers or None, timeout=_as_timeout(timeout), transport=transport)
property
¶
is_cloud: bool
Check whether connection uses cloud mode.
Returns True for cloud connection_type.
property
¶
is_local: bool
Check whether connection uses local mode.
Returns True for local connection_type.
async
¶
usb_reboot(*, raise_on_error: bool = False) -> bool
Attempt to reboot the device via USB.
async
¶
usb_reset(*, raise_on_error: bool = False) -> bool
Alias for usb_reboot().
Provided for callers that prefer "reset" naming.
method_compatibility(method_name: str) -> versioning.MethodCompatibility | None
Return declarative OpenAPI compatibility metadata for a client method.
async
¶
is_local_available() -> bool
Check local API reachability on base_url.
Returns True when /api/version responds without network errors.
async
¶
api_request(method: str, path: str, *, params: dict[str, Any] | None = None, headers: dict[str, str] | None = None, session_id: str | None = None, application_name: str | None = None, json_payload: JsonType | None = None, data: bytes | AsyncIterable[bytes] | None = None, expect_bytes: bool = False, allow_text: bool = False, timeout: float | Timeout | None = None) -> JsonType | bytes | str
Execute a raw async API request through the current client session.
Advanced callers can control path, params, headers, body, and request context without creating a separate HTTP client.
prepare_request(method: str, path: str, *, params: dict[str, Any] | None = None, headers: dict[str, str] | None = None, session_id: str | None = None, application_name: str | None = None, json_payload: JsonType | None = None, data: bytes | AsyncIterable[bytes] | None = None, expect_bytes: bool = False, allow_text: bool = False, timeout: float | Timeout | None = None) -> PreparedRequest
Build a prepared async request without executing network I/O.
External integrations can inspect the prepared payload, route it to
custom transports, or execute later via execute_prepared_request.
Prepared request should be executed with an async executor.
async
¶
execute_prepared_request(prepared: PreparedRequest, *, client: AsyncClient | None = None) -> JsonType | bytes | str
Execute a previously prepared async request.
By default the current httpx2.AsyncClient is used. Callers may inject
a custom client while preserving error mapping. Prepared streaming
content is single-use and should be regenerated for repeated executions.
async
¶
stream_status_ws(*, enable: bool = True, decode_protobuf: bool = True) -> AsyncIterator[dict[str, Any] | bytes | str]
Open /api/status/ws and yield status updates.
When decode_protobuf=True, binary frames are decoded using
BSB_State.State protobuf schema from bsb-protobuf and converted into
dictionaries with original proto field names.
async
¶
smart_home_pairing() -> types.SmartHomePairingInfo
Fetch smart home pairing status via GET /api/smart_home/pairing.
async
¶
smart_home_pairing_start() -> types.SmartHomePairingPayload
Start smart home pairing via POST /api/smart_home/pairing.
async
¶
smart_home_pairing_stop() -> types.SuccessResponse
Stop smart home pairing via DELETE /api/smart_home/pairing.
async
¶
smart_home_switch() -> types.SmartHomeSwitchState
Fetch smart home switch state via GET /api/smart_home/switch.
async
¶
smart_home_switch_set(state: bool, *, startup: Literal['off', 'on', 'toggle', 'last'] | None = None) -> types.SuccessResponse
Set smart home switch state via POST /api/smart_home/switch.
async
¶
wifi_enable() -> types.SuccessResponse
Removed from the device API.
No supported firmware serves POST /api/wifi/enable, so calling this raises
BusyBarRemovedEndpointError. Use wifi_connect() / wifi_disconnect() instead.
async
¶
wifi_disable() -> types.SuccessResponse
Removed from the device API.
No supported firmware serves POST /api/wifi/disable, so calling this raises
BusyBarRemovedEndpointError. Use wifi_connect() / wifi_disconnect() instead.
async
¶
wifi_connect(config: ConnectRequestConfig | dict[str, Any]) -> types.SuccessResponse
async
¶
wifi_networks() -> types.NetworkResponse
Scan for nearby networks via GET /api/wifi/networks.
The device cannot scan while it is associated: doing so returns
400 "Scan not possible when connected" as a BusyBarAPIError.
Disconnect first with wifi_disconnect(), or skip the scan and pass
the SSID to wifi_connect() directly.
async
¶
audio_play(*, path: str | None = None, stock_path: str | None = None, payload: AudioPlayRequest | dict[str, Any] | None = None, **request_kwargs: Unpack[RequestKwargs]) -> types.SuccessResponse
Play audio through async POST /api/audio/play.
The endpoint payload may reference either an uploaded asset path or a
stock path. Explicit path and stock_path keyword arguments override
the same keys provided in payload. Request context such as
application_name and session_id is accepted through request kwargs;
use api_request for fully custom bodies.
async
¶
audio_stop() -> types.SuccessResponse
Stop audio through async DELETE /api/audio/play.
Uses API-like naming for callers that mirror firmware endpoints.
async
¶
display_draw(display_data: DisplayElements | dict[str, Any], *, clear_before_draw: bool = False, sanitize_text: bool = False, **request_kwargs: Unpack[RequestKwargs]) -> types.SuccessResponse
async
¶
display(display_data: DisplayElements | dict[str, Any], *, clear_before_draw: bool = False, sanitize_text: bool = False, audio_payload: AudioPlayRequest | dict[str, Any] | None = None, **request_kwargs: Unpack[RequestKwargs]) -> types.SuccessResponse
Render display content and optionally play audio after draw.
Operations are sequential, not atomic: clear may succeed before draw fails, and audio failure may occur after display content is visible. Exceptions include the failed endpoint path for diagnostics.
async
¶
display_clear(**request_kwargs: Unpack[RequestKwargs]) -> types.SuccessResponse
Clear display content through async DELETE /api/display/draw.
Uses API-like naming for callers that mirror firmware endpoints.
async
¶
display_brightness_set(value: BrightnessValue) -> types.SuccessResponse
async
¶
screen(display_id: DisplaySpecLike) -> bytes
Fetch a single display frame via GET /api/screen.
Returns RGB bytes, three per pixel: 3456 for the front display (72x16) and 38400 for the back (160x80, sent L4-packed and expanded to grey triples here). The device sends colour as BGR and it is swapped on the way out.
The response body is base64-encoded, uncompressed framebuffer bytes
(the Content-Type: image/bmp header is misleading, there is no
real BMP header).
async
¶
frame(display_id: DisplaySpecLike) -> Frame
Fetch a display frame as a Frame via GET /api/screen.
The same bytes as screen(), with the geometry and the display
attached, so callers can read pixels, rows or a PNG without tracking
the layout themselves.
async
¶
assets_upload(application_name: str, filename: str, data: bytes, *, timeout: float | Timeout | None = ASSET_UPLOAD_TIMEOUT) -> types.SuccessResponse
Upload an asset file for the given application.
Uses a longer default timeout to tolerate large payload uploads.
async
¶
storage_write(path: str, data: bytes, *, timeout: float | None = 60.0, progress_callback: Callable[[int, int], None] | None = None, chunk_size: int = 64 * 1024) -> types.SuccessResponse
async
¶
storage_mkdir(path: str) -> types.SuccessResponse
Create a storage directory via POST /api/storage/mkdir.
async
¶
storage_rename(old_path: str, new_path: str) -> types.SuccessResponse
Rename a storage entry via POST /api/storage/rename.
async
¶
transport() -> types.NetworkInterfaceInfo
Fetch active network transport via GET /api/transport.
async
¶
status_device() -> types.StatusDevice
Fetch device manufacturing status via GET /api/status/device.
async
¶
status_firmware() -> types.StatusFirmware
Fetch firmware status via GET /api/status/firmware.
async
¶
status_system() -> types.StatusSystem
Fetch runtime status via GET /api/status/system.
async
¶
status_power() -> types.StatusPower
Fetch power status via GET /api/status/power.
async
¶
log_dump(filename: str | None = None) -> types.LogDumpResponse
Dump the in-memory device log buffer to a storage file.
filename is a bare name without a path or extension, matching
^[a-zA-Z0-9_-]+$ on firmware OpenAPI 25.0.0+; the device appends its
own extension and storage path. When omitted, the device picks a
default file.
Breaking change: prior to 25.0.0 this method accepted path= (a full
device-side path). That parameter has been removed rather than
aliased, since the two contracts are not translatable (a full path
never matches the new filename pattern). Callers targeting firmware
older than 25.0.0 should pin an older busylib release instead of
adapting call sites.
async
¶
update(firmware_data: bytes) -> types.SuccessResponse
Upload firmware update TAR and initiate update.
async
¶
update_check() -> types.SuccessResponse
Start asynchronous firmware update check.
async
¶
update_status() -> types.UpdateStatus
Get firmware update status with progress information.
async
¶
update_changelog(version: str) -> types.UpdateChangelogResponse
Fetch update changelog for a specific version.
async
¶
update_install(version: str) -> types.SuccessResponse
Start firmware update installation by version.
async
¶
update_abort_download() -> types.SuccessResponse
Abort an ongoing firmware download.
async
¶
update_autoupdate() -> types.AutoupdateSettings
Fetch autoupdate settings via GET /api/update/autoupdate.
async
¶
update_autoupdate_set(settings: AutoupdateSettings | dict[str, object]) -> types.SuccessResponse
Set autoupdate settings via POST /api/update/autoupdate.
async
¶
time_timezone_info() -> types.TimezoneInfo
Fetch current device timezone via GET /api/time/timezone.
async
¶
time_timezone_list() -> types.TimezoneListResponse
Fetch supported device timezones via GET /api/time/tzlist.
async
¶
time_timestamp(timestamp: str) -> types.SuccessResponse
Set device time via POST /api/time/timestamp.
async
¶
time_timezone(timezone: str) -> types.SuccessResponse
Set device timezone via POST /api/time/timezone.
async
¶
busy_snapshot() -> types.BusySnapshot
Fetch busy snapshot via GET /api/busy/snapshot.
async
¶
busy_snapshot_set(snapshot: BusySnapshot) -> types.SuccessResponse
Set busy snapshot via PUT /api/busy/snapshot.
async
¶
busy_profile(slot: BusyProfileSlot) -> types.BusyProfile
Fetch a busy profile slot via GET /api/busy/profiles/{slot}.
async
¶
busy_profile_set(slot: BusyProfileSlot, profile: BusyProfile | dict[str, object]) -> types.SuccessResponse
Set a busy profile slot via PUT /api/busy/profiles/{slot}.
async
¶
account_unlink() -> types.SuccessResponse
Unlink the device from the account via DELETE /api/account.
async
¶
account_link() -> types.AccountLink
Request an account link code via POST /api/account/link.
async
¶
account_info() -> types.AccountInfo
Fetch linked account info via GET /api/account/info.
async
¶
account_status() -> types.AccountState
Fetch MQTT status info via GET /api/account/status.
async
¶
account_backend() -> types.AccountBackend
Fetch MQTT backend settings via GET /api/account/backend.
async
¶
account_backend_set(backend: AccountBackend | dict[str, object]) -> types.SuccessResponse
Set MQTT backend settings via PUT /api/account/backend.
async
¶
account_profile() -> types.AccountProfile
Removed from the device API.
No supported firmware serves GET /api/account/profile, so calling this raises
BusyBarRemovedEndpointError. Use account_backend() instead.
async
¶
account_profile_set(profile: AccountProfileName, custom_url: str | None = None) -> types.SuccessResponse
Removed from the device API.
No supported firmware serves POST /api/account/profile, so calling this raises
BusyBarRemovedEndpointError. Use account_backend_set() instead.
async
¶
access_set(mode: HttpAccessMode, key: str) -> types.SuccessResponse
Set HTTP access mode via POST /api/access.
async
¶
access_tokens_list() -> types.AccessTokensInfo
List issued access tokens via GET /api/access/tokens.
The secret itself is returned only when a token is minted, so every
token here has token set to None.
Experimental: needs firmware from busy-app/busybar-firmware#886, which is not released yet, so it answers 404 on every released firmware.
async
¶
access_token_mint(name: str) -> types.AccessToken
Issue a new access token via POST /api/access/tokens.
This is the only time the device discloses the secret; it cannot be read back afterwards.
Experimental: needs firmware from busy-app/busybar-firmware#886, which is not released yet, so it answers 404 on every released firmware.
async
¶
access_tokens_delete_all() -> types.SuccessResponse
Revoke every access token via DELETE /api/access/tokens.
Experimental: needs firmware from busy-app/busybar-firmware#886, which is not released yet, so it answers 404 on every released firmware.
async
¶
access_tokens_revoke(short_id: str) -> types.SuccessResponse
Revoke one access token via DELETE /api/access/tokens/{short_id}.
Experimental: needs firmware from busy-app/busybar-firmware#886, which is not released yet, so it answers 404 on every released firmware.
Prepared requests¶
Bases: BaseModel
Prepared low-level request ready for execution by HTTP clients.
This object stores normalized request attributes after payload encoding. Callers can execute it with built-in client transport or pass fields to an external request executor. Serialization is not guaranteed because timeout and streaming content may contain runtime-only objects.
Streaming content based on iterables/generators is single-use and should not be reused across multiple executions.
class-attribute
instance-attribute
¶
headers: dict[str, str] | None = Field(default=None, repr=False)
class-attribute
instance-attribute
¶
content: bytes | Iterable[bytes] | AsyncIterable[bytes] | None = Field(default=None, repr=False)
class-attribute
instance-attribute
¶
model_config = ConfigDict(arbitrary_types_allowed=True)