kiarina-lib-firebase-firestore 2.20.0

Read-only Cloud Firestore REST client library for kiarina namespace

kiarina-lib-firebase-firestore

English | 日本語

[!NOTE] What is this? An asynchronous read-only package for retrieving documents from Cloud Firestore with a Firebase ID token.

Dependencies

Package	Version	License
HTTPX	>=0.28.1	BSD-3-Clause
Pydantic	>=2.10.6	MIT
Pydantic Settings	>=2.10.1	MIT
pydantic-settings-manager	>=3.2.0	MIT

Installation

pip install kiarina-lib-firebase-firestore

Features

• Retrieving a Document Retrieves the document at a path through the Firestore REST API.
• Listing Documents Lists documents in a collection with pagination.
• Decoding Firestore Values Converts Firestore typed values (such as integerValue) into Python values.
• Read Only by Design Provides no write APIs. Writes are expected to go through the server side (such as an API server).
• Configuring the Client Configures the endpoint and timeout through environment variables or pydantic-settings-manager.

Retrieving a Document

Pass a Firebase ID token obtained through TokenManager (kiarina-lib-firebase) or similar, and the document path.

from kiarina.lib.firebase import TokenManager
from kiarina.lib.firebase_firestore import get_document

token_manager = TokenManager(
    api_key="firebase-web-api-key",
    refresh_token="firebase-refresh-token",
)

snapshot = await get_document(
    "your-project-id",
    "users/user_1/posts/post_1",
    await token_manager.get_id_token(),
)

if snapshot is not None:
    print(snapshot.id, snapshot.fields)

Returns None when the document does not exist. Raises httpx.HTTPStatusError (403) when denied by security rules.

Listing Documents

list_documents lists documents in a collection. By default, documents are returned in document-name order.

from kiarina.lib.firebase_firestore import list_documents

result = await list_documents(
    "your-project-id",
    "users/user_1/posts",
    id_token,
    page_size=100,
)

for snapshot in result.documents:
    print(snapshot.id, snapshot.fields)

if result.next_page_token is not None:
    next_page = await list_documents(
        "your-project-id",
        "users/user_1/posts",
        id_token,
        page_size=100,
        page_token=result.next_page_token,
    )

Configuring the Client

Settings are managed by the single-configuration settings_manager.

kiarina.lib.firebase_firestore:
  base_url: https://firestore.googleapis.com
  timeout: 30.0

Load the settings at application startup.

import yaml
from pydantic_settings_manager import load_user_configs

from kiarina.lib.firebase_firestore 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 directly, assign values to settings_manager.user_config.

from kiarina.lib.firebase_firestore import settings_manager

settings_manager.user_config = {
    "base_url": "http://localhost:8080",
    "timeout": 30.0,
}

Environment variables are also supported. Point base_url at a Firestore emulator for local testing.

export KIARINA_LIB_FIREBASE_FIRESTORE_BASE_URL=http://localhost:8080
export KIARINA_LIB_FIREBASE_FIRESTORE_TIMEOUT=30.0

API Reference

kiarina.lib.firebase_firestore

from kiarina.lib.firebase_firestore import (
    DocumentList,
    DocumentSnapshot,
    FirestoreSettings,
    get_document,
    list_documents,
    settings_manager,
)

get_document

async def get_document(
    project_id: str,
    path: str,
    id_token: str,
    *,
    database_id: str = "(default)",
) -> DocumentSnapshot | None: ...

Retrieves the document at the specified path.

Parameters

• project_id (str): Google Cloud project ID
• path (str): Document path (e.g. "users/user_1/posts/post_1")
• id_token (str): Firebase ID token
• database_id (str): Database ID. Defaults to "(default)"

Returns

• DocumentSnapshot | None: The document, or None when it does not exist

Raises

• httpx.HTTPStatusError: When the HTTP response indicates an error (except 404)
• httpx.HTTPError: When communication fails

list_documents

async def list_documents(
    project_id: str,
    collection_path: str,
    id_token: str,
    *,
    database_id: str = "(default)",
    page_size: int | None = None,
    page_token: str | None = None,
    order_by: str | None = None,
) -> DocumentList: ...

Lists documents in a collection.

Parameters

• project_id (str): Google Cloud project ID
• collection_path (str): Collection path (e.g. "users/user_1/posts")
• id_token (str): Firebase ID token
• database_id (str): Database ID. Defaults to "(default)"
• page_size (int | None): Maximum number of documents per page
• page_token (str | None): The next_page_token from the previous page
• order_by (str | None): Sort order (e.g. "createTime desc")

Returns

• DocumentList: A page of documents

Raises

• httpx.HTTPStatusError: When the HTTP response indicates an error
• httpx.HTTPError: When communication fails

DocumentSnapshot

@dataclass
class DocumentSnapshot:
    name: str
    fields: dict[str, Any]
    create_time: datetime
    update_time: datetime

    @property
    def path(self) -> str: ...

    @property
    def id(self) -> str: ...

A document retrieved from Cloud Firestore.

Fields

• name (str): Full resource name of the document
• fields (dict[str, Any]): Fields converted into Python values
• create_time (datetime): Creation time
• update_time (datetime): Update time

Properties

• path (str): Path relative to the database root (e.g. "users/user_1/posts/post_1")
• id (str): Document ID (the last segment of the path)

Field values are converted as follows.

Firestore	Python
nullValue	None
booleanValue	bool
integerValue	int
doubleValue	float
timestampValue	datetime
stringValue	str
bytesValue	bytes
referenceValue	str (resource name)
geoPointValue	dict (latitude / longitude)
arrayValue	list
mapValue	dict

DocumentList

@dataclass
class DocumentList:
    documents: list[DocumentSnapshot]
    next_page_token: str | None

A page of documents listed from a collection.

Fields

• documents (list[DocumentSnapshot]): Documents in this page
• next_page_token (str | None): Token for retrieving the next page. None on the last page

FirestoreSettings

class FirestoreSettings(BaseSettings):
    base_url: str = "https://firestore.googleapis.com"
    timeout: float = 30.0

Settings for the Firestore REST client.

Fields

• base_url (str): Base URL of the Firestore REST API. Point this at a Firestore emulator for local testing
• timeout (float): HTTP request timeout in seconds

settings_manager

settings_manager = SettingsManager(FirestoreSettings)

The SettingsManager for FirestoreSettings.

License

MIT License - See LICENSE for details.