kiarina-lib-firebase-rtdb

Dependencies

Installation

pip install kiarina-lib-firebase-rtdb

Features

• Retrieving Data Retrieves data at a path through the Firebase Realtime Database REST API.
• Querying Data Orders, limits and ranges the result with RTDBQuery, which encodes the REST query parameters.
• Updating Data Writes a multi-path update, and deletes keys by sending None.
• Watching Data Changes Yields the current value of a path each time it changes, kept in sync from the Server-Sent Events stream.
• Reading Your Own Writes Shares an RTDBMirror between watch_data and update_data, so a successful update is visible locally before the stream echoes it back.
• Recovering the Stream Refreshes the ID token after authentication revocation and reconnects with exponential backoff after network errors and token refresh failures.
• Stopping the Stream Stops a watch with an asyncio.Event.
• Resolving the Token Passes a token explicitly, or uses the token manager of the named kiarina.lib.firebase settings.
• Configuring Retries Configures retry intervals through environment variables or pydantic-settings-manager.

Retrieving Data

Get a token from a TokenManager and specify a database path. create_token_manager reads the API key and the token file from the kiarina.lib.firebase settings.

from kiarina.lib.firebase import create_token_manager
from kiarina.lib.firebase_rtdb import get_data

token_manager = create_token_manager()

data = await get_data(
    "https://your-project-default-rtdb.firebaseio.com",
    "/agents/state",
    token=await token_manager.get_token(),
)

Querying Data

RTDBQuery builds the REST query parameters and JSON-encodes their values, which the REST API requires. Ordering by $key needs no index and is chronological when keys are ULIDs.

from kiarina.lib.firebase_rtdb import RTDBQuery, get_data

data = await get_data(
    "https://your-project-default-rtdb.firebaseio.com",
    "/agents/messages",
    query=RTDBQuery(order_by="$key", limit_to_last=5),
    token=await token_manager.get_token(),
)

Pass start_after to fetch only the entries added after the last key already seen.

query = RTDBQuery(order_by="$key", start_after="01ABCDEF...")

shallow truncates every value to true and returns the keys alone. The REST API rejects it together with any other parameter, so RTDBQuery raises a validation error for that combination.

keys = await get_data(
    "https://your-project-default-rtdb.firebaseio.com",
    "/agents/messages",
    query=RTDBQuery(shallow=True),
    token=await token_manager.get_token(),
)

Updating Data

update_data sends a multi-path update. Keys are paths relative to the given path, and a None value deletes the key.

from kiarina.lib.firebase_rtdb import update_data

await update_data(
    "https://your-project-default-rtdb.firebaseio.com",
    "/agents/messages",
    {"01ABCDEF.../read": True, "01OLDEST...": None},
    token=await token_manager.get_token(),
)

Watching Data Changes

watch_data yields the whole value at the path, in the same shape get_data returns, each time it changes. It applies the stream's put and patch events to a local copy, so deletions are reflected too: a removed child disappears from the next value, and the value is None while the path does not exist.

from kiarina.lib.firebase_rtdb import watch_data

async for entries in watch_data(
    "https://your-project-default-rtdb.firebaseio.com",
    "/agents/entries",
    token_manager=token_manager,
):
    unread = [key for key, entry in (entries or {}).items() if not entry.get("read")]

The first value arrives once Firebase sends the initial snapshot. A value equal to the previous one is not yielded again. Each yielded value is an independent copy, so the caller may modify it.

Reading Your Own Writes

Pass the same RTDBMirror to watch_data and update_data. watch_data binds the mirror to the watched path and keeps it in sync. update_data applies the update to it before sending, so mirror.value reflects the write before the stream echoes it back, even when the echo arrives before the response. The echo then changes nothing, so watch_data does not yield it again. If the update fails, the parts it changed are rolled back.

from kiarina.lib.firebase_rtdb import RTDBMirror, update_data, watch_data

mirror = RTDBMirror()

async for entries in watch_data(DATABASE_URL, "/agents/entries", mirror=mirror):
    ...

# Elsewhere, while the watch runs
await update_data(DATABASE_URL, "/agents/entries", {"01ABC/read": True}, mirror=mirror)
entries = mirror.value  # already has read=True

Only the part of an update under the mirrored path is applied. An update before the first snapshot is not applied, because the snapshot replaces the value anyway.

Recovering the Stream

When authentication is revoked, it calls TokenManager.refresh() and reconnects. An ID token lives for one hour, so this reconnect happens periodically for as long as the watch runs. Right after a reconnect Firebase sends the whole path again, so changes made while disconnected are reflected in the next value.

Network errors and transient token refresh failures use the configured exponential backoff. Errors that retrying cannot recover from, such as an invalidated refresh token, are propagated to the caller.

Stopping the Stream

Setting stop_event ends the watch when the stream next receives data. Cancel the watch task when an immediate stop is required.

import asyncio

from kiarina.lib.firebase_rtdb import watch_data

stop_event = asyncio.Event()

async for value in watch_data(
    "https://your-project-default-rtdb.firebaseio.com",
    "/agents/state",
    stop_event=stop_event,
    token_manager=token_manager,
):
    print(value)
    if value == "stop":
        stop_event.set()

Resolving the Token

Omitting token and token_manager uses the TokenManager of the kiarina.lib.firebase settings named by firebase_settings_key.

kiarina.lib.firebase:
  configs:
    production:
      api_key: production-api-key
      token_file_path: ~/.config/your-app/token.json

kiarina.lib.firebase_rtdb:
  firebase_settings_key: production

data = await get_data(
    "https://your-project-default-rtdb.firebaseio.com",
    "/agents/state",
)

token_manager_registry builds the token manager from those settings. Register an instance under the same key to use a different TokenStore.

Omitting firebase_settings_key uses the default of token_manager_registry, which is the kiarina.lib.firebase settings that its settings_manager resolves.

Configuring Retries

Retry settings use a single-mode settings_manager.

kiarina.lib.firebase_rtdb:
  max_retry_delay: 60.0
  initial_retry_delay: 1.0
  retry_delay_multiplier: 2.0

Load the settings when the application starts.

import yaml
from pydantic_settings_manager import load_user_configs

from kiarina.lib.firebase_rtdb import settings_manager

with open("config.yaml", encoding="utf-8") as file:
    load_user_configs(yaml.safe_load(file) or {})

settings = settings_manager.get_settings()

To configure only this package, assign the values directly to settings_manager.user_config.

from kiarina.lib.firebase_rtdb import settings_manager

settings_manager.user_config = {
    "max_retry_delay": 60.0,
    "initial_retry_delay": 1.0,
    "retry_delay_multiplier": 2.0,
}

The same values are available as environment variables.

export KIARINA_LIB_FIREBASE_RTDB_MAX_RETRY_DELAY=60.0
export KIARINA_LIB_FIREBASE_RTDB_INITIAL_RETRY_DELAY=1.0
export KIARINA_LIB_FIREBASE_RTDB_RETRY_DELAY_MULTIPLIER=2.0

API Reference

kiarina.lib.firebase_rtdb

from kiarina.lib.firebase_rtdb import (
    RTDBMirror,
    RTDBQuery,
    RTDBSettings,
    RTDBStreamCancelledError,
    get_data,
    settings_manager,
    update_data,
    watch_data,
)

get_data

async def get_data(
    database_url: str,
    path: str,
    *,
    query: RTDBQuery | None = None,
    token: Token | None = None,
) -> Any: ...

Retrieves JSON data at the specified path.

Parameters

• database_url (str): Firebase Realtime Database URL
• path (str): Path of the data to retrieve
• query (RTDBQuery | None): Query parameters appended to the request
• token (Token | None): Firebase token set. Resolved from token_manager_registry when omitted

Returns

• Any: JSON value from the response

Raises

• ValueError: The token is omitted and token_manager_registry cannot resolve a TokenManager
• httpx.HTTPStatusError: The HTTP response indicates an error. Its message, request, and response omit the auth query parameter that carries the ID token
• httpx.HTTPError: The request fails

update_data

async def update_data(
    database_url: str,
    path: str,
    values: Mapping[str, Any],
    *,
    token: Token | None = None,
    mirror: RTDBMirror | None = None,
) -> Any: ...

Applies a multi-path update at the specified path.

Parameters

• database_url (str): Firebase Realtime Database URL
• path (str): Path the update is applied to
• values (Mapping[str, Any]): Keys relative to path and their new values. None deletes the key
• token (Token | None): Firebase token set. Resolved from token_manager_registry when omitted
• mirror (RTDBMirror | None): Mirror to apply the update to before sending. Only the part under the mirrored path is applied, and it is rolled back if the update fails

Returns

• Any: JSON value from the response

Raises

• ValueError: The token is omitted and token_manager_registry cannot resolve a TokenManager
• httpx.HTTPStatusError: The HTTP response indicates an error. Its message, request, and response omit the auth query parameter that carries the ID token
• httpx.HTTPError: The request fails

watch_data

async def watch_data(
    database_url: str,
    path: str,
    *,
    stop_event: asyncio.Event | None = None,
    token_manager: TokenManager | None = None,
    mirror: RTDBMirror | None = None,
) -> AsyncIterator[Any]: ...

Watches the specified path and yields its whole value each time it changes, kept in sync from the Firebase SSE stream.

Parameters

• database_url (str): Firebase Realtime Database URL
• path (str): Path of the data to watch
• stop_event (asyncio.Event | None): Event that requests the watch to stop
• token_manager (TokenManager | None): Instance that manages the token set. Resolved from token_manager_registry when omitted
• mirror (RTDBMirror | None): Mirror to keep in sync with the stream. A new one is used when omitted

Yields

• Any: The value at the path, in the same shape get_data returns, or None when the path does not exist. The first value is yielded after the initial snapshot arrives, a value equal to the previous one is skipped, and each value is an independent copy

Raises

• ValueError: The token is omitted and token_manager_registry cannot resolve a TokenManager, or mirror is already bound to another path
• RTDBStreamCancelledError: Firebase cancels the stream
• InvalidRefreshTokenError: The refresh token is no longer usable
• FirebaseAPIError: Token refresh fails with an error that retrying cannot recover from

Network errors, HTTP error responses, and transient token refresh failures are retried internally. HTTP error responses are logged without the ID token. Other unexpected exceptions are propagated to the caller.

RTDBMirror

class RTDBMirror:
    def __init__(self) -> None: ...

    @property
    def value(self) -> Any: ...

A local copy of one Firebase Realtime Database path. watch_data binds it to the watched path and keeps it in sync, and update_data applies updates to it before sending and rolls them back on failure.

Properties

• value (Any): The current value, or None before the first snapshot and while the path does not exist. Each access returns an independent copy

RTDBQuery

class RTDBQuery(BaseModel):
    order_by: str | None = None
    limit_to_first: int | None = None
    limit_to_last: int | None = None
    start_at: QueryValue | None = None
    start_after: QueryValue | None = None
    end_at: QueryValue | None = None
    end_before: QueryValue | None = None
    equal_to: QueryValue | None = None
    shallow: bool = False

Query parameters for the Firebase Realtime Database REST API. QueryValue is str | bool | int | float.

Fields

• order_by (str | None): Child key to order by, or "$key", "$value" or "$priority"
• limit_to_first (int | None): Number of items to take from the beginning of the ordered result
• limit_to_last (int | None): Number of items to take from the end of the ordered result
• start_at (QueryValue | None): Inclusive lower bound of the ordered result
• start_after (QueryValue | None): Exclusive lower bound of the ordered result
• end_at (QueryValue | None): Inclusive upper bound of the ordered result
• end_before (QueryValue | None): Exclusive upper bound of the ordered result
• equal_to (QueryValue | None): Exact value the ordered child must match
• shallow (bool): Truncate each value to true

Methods

• to_params() -> dict[str, str]: Returns the REST query parameters with JSON-encoded values

Raises

• ValidationError: shallow is combined with another parameter, a filter is used without order_by, or mutually exclusive parameters are set together

RTDBSettings

class RTDBSettings(BaseSettings):
    firebase_settings_key: str | None = None
    max_retry_delay: float = 60.0
    initial_retry_delay: float = 1.0
    retry_delay_multiplier: float = 2.0

Settings used when resolving the token and reconnecting a stream.

Fields

• firebase_settings_key (str | None): Key of the kiarina.lib.firebase settings whose TokenManager is used when no token is passed. An alias of kiarina.lib.firebase is also accepted. The default of token_manager_registry is used when this is not set
• max_retry_delay (float): Maximum retry interval in seconds
• initial_retry_delay (float): Initial retry interval in seconds
• retry_delay_multiplier (float): Value multiplied by the retry interval after a network error

settings_manager

settings_manager: SettingsManager[RTDBSettings]

Manages a single RTDBSettings configuration.

RTDBStreamCancelledError

class RTDBStreamCancelledError(Exception): ...

Indicates that Firebase cancelled the SSE stream.