Firedantic Extras

Add-on utilities for Firedantic — the async-native Pydantic + Firestore ODM.

Overview

Firedantic Extras is a companion library that provides higher-level utilities built on top of Firedantic models. Each module solves a specific, recurring problem that arises when using Firedantic in production:

Sync and async

Every helper that talks to Firestore comes in two flavours. The plain name works with firedantic.Model (sync); the async_ / Async name works with firedantic.AsyncModel and must be awaited. Options, return types and behaviour are identical.

build_prefix_filters, CursorPage, SyncResult and the bigquery.schema module never touch Firestore and are shared. The Flask adapter is sync-only (Flask is); the FastAPI adapter's PaginationParams works with either — see the FastAPI section.

Don't call a sync helper from inside a running event loop (an async def FastAPI route, for example): it blocks the loop. Use the async flavour there.

cursor_pagination — Cursor-Based Pagination

Efficient, stable, bidirectional cursor pagination for any Firedantic model. Works independently of any web framework — use it from Flask, FastAPI, background workers, or anywhere else.

Why?

Model.find() returns every document in the collection. For large collections this becomes slow and expensive. cursor_paginate() fetches only one page at a time using Firestore's start_after cursor, so response time stays constant no matter how many documents exist.

Key design decisions:

• Document ID as cursor — stable, type-safe, no field serialization needed.
• __name__ tiebreaker — a final order_by("__name__") prevents silent deduplication at page boundaries when the primary sort field has duplicates.
• Reversed sort for prev — backward pagination reverses all sort directions and uses start_after (instead of end_before + limit_to_last), which is fully supported by both the Firestore SDK and the emulator.
• Sentinel row — requests limit + 1 rows to determine has_next / has_prev without an extra COUNT query.

Quick Start

from firedantic import Model
from firedantic_extras import cursor_paginate

class Product(Model):
    __collection__ = "products"
    name: str
    price: float
    category: str

# --- Page 1 (forward, no cursor) ---
page = cursor_paginate(Product, limit=20, order_by="name")
for p in page.items:
    print(p.name)

print(page.has_next)    # True  — more items exist
print(page.has_prev)    # False — we're on the first page

# --- Page 2 (next) ---
page2 = cursor_paginate(
    Product,
    limit=20,
    order_by="name",
    cursor=page.next_cursor,
    direction="next",
)

# --- Back to Page 1 (prev) ---
page1_again = cursor_paginate(
    Product,
    limit=20,
    order_by="name",
    cursor=page2.prev_cursor,
    direction="prev",
)

# --- Last page (no cursor, going backward) ---
last_page = cursor_paginate(Product, limit=20, order_by="name", direction="prev")
print(last_page.has_next)  # False — nothing after this
print(last_page.has_prev)  # True  — earlier pages exist

Filtering

Pass a filter_ dict using the same format as Model.find():

# Equality filter
page = cursor_paginate(
    Product,
    limit=20,
    order_by="name",
    filter_={"category": "electronics"},
)

# Comparison operators
page = cursor_paginate(
    Product,
    limit=20,
    order_by=[("price", "ASCENDING"), ("name", "ASCENDING")],
    filter_={"price": {">=": 10.0, "<": 100.0}},
)

# OR filters (firedantic 0.18+): top-level keys are ANDed, $or takes a list
# of filter dicts and nests with $and
from firedantic import operators as op

page = cursor_paginate(
    Product,
    limit=20,
    order_by="name",
    filter_={
        "in_stock": True,
        op.OR: [{"category": "electronics"}, {op.AND: [{"category": "toys"}, {"price": {op.LT: 20}}]}],
    },
)

The filter dict is handed straight to firedantic's find(), so anything it accepts works here, in count_model() and in the Flask/FastAPI adapters.

Compound Sort

Pass a list of (field, direction) tuples for multi-field ordering:

from google.cloud.firestore_v1 import ASCENDING, DESCENDING

page = cursor_paginate(
    Product,
    limit=20,
    order_by=[("category", ASCENDING), ("price", DESCENDING)],
)

Include Total Count

page = cursor_paginate(Product, limit=20, order_by="name", include_total=True)
print(page.total)  # e.g. 4231 — one extra server-side COUNT aggregation

Excluding Null Sort Fields

Firestore sorts documents where the sort field is null before all non-null values in ascending order (and after, in descending order). When a collection has a mix of documents with and without an optional field populated, sorting by that field surfaces a run of useless null-valued documents before any real data appears.

# Sorting by an optional field — skip null/missing values
page = cursor_paginate(
    Product,
    limit=100,
    order_by="discontinued_at",
    exclude_null_sort_field=True,
)

This adds a != None filter on the primary sort field (the first entry in order_by), which Firestore excludes both null-valued and missing-field documents for — no need to guess a type-specific sentinel value. Requires order_by to be set, and raises ValueError if filter_ already has an entry for that field.

Falling Back on a Missing Composite Index

If you use Firestore's "constrained path" index strategy — hand-defining only the composite indexes you actually need instead of every filter+sort permutation — a filter + order_by combination without a matching index raises FailedPrecondition. fallback_order_by retries once with a safe sort instead of making every caller write the same try/except:

page = cursor_paginate(
    Product,
    limit=100,
    filter_=filter_,
    order_by=[("barcode", "ASCENDING")],
    fallback_order_by=[("updated_at", "DESCENDING")],
)
if page.used_fallback:
    flash("Sorting by barcode isn't available with this filter — showing most recently updated instead.")

The cursor and direction are reset (cursor=None, direction="next") on fallback, since a cursor encoded for one sort order is meaningless under another. With no fallback_order_by, FailedPrecondition propagates as usual — this is purely opt-in.

API Reference

def cursor_paginate(
    model_class: type[BareModel],
    *,
    limit: int,
    order_by: str | list[str | tuple[str, str]] | None = None,
    cursor: str | None = None,
    direction: Literal["next", "prev"] = "next",
    filter_: FilterDict | None = None,
    include_total: bool = False,
    exclude_null_sort_field: bool = False,
    fallback_order_by: str | list[str | tuple[str, str]] | None = None,
) -> CursorPage[BareModel]:
    ...

@dataclass
class CursorPage(Generic[ModelT]):
    items: list[ModelT]       # hydrated model instances for this page
    has_next: bool            # True if a next page exists (going forward)
    has_prev: bool            # True if a previous page exists (going backward)
    next_cursor: str | None   # pass as cursor + direction="next" to advance
    prev_cursor: str | None   # pass as cursor + direction="prev" to go back
    total: int | None         # total doc count, only set when include_total=True
    used_fallback: bool       # True if this page was fetched with fallback_order_by

query — Count and Prefix Search

count_model — Server-Side Aggregation

Get the number of documents matching a filter without fetching any data:

from firedantic_extras import count_model

# Count all documents in a collection
total = count_model(Product)

# Count with a filter
electronics_count = count_model(Product, filter_={"category": "electronics"})

Uses Firestore's native COUNT aggregation — no documents are transferred. A thin wrapper over firedantic's Model.count(), kept so the sync and async variants share one import and the filter_ convention (including $or / $and) matches cursor_paginate.

def count_model(
    model_class: type[BareModel],
    *,
    filter_: FilterDict | None = None,
) -> int:
    ...

build_prefix_filters — Prefix-Range Search

Generate a pair of Firedantic-compatible filters that implement a prefix search using Firestore's range query pattern:

from firedantic_extras import build_prefix_filters, cursor_paginate

# Find all products whose name starts with "lap"
filters = build_prefix_filters("name", "lap")
# Returns: {"name": {">=": "lap", "<": "lap\uf8ff"}}

page = cursor_paginate(
    Product,
    limit=20,
    order_by="name",
    filter_=filters,
)

The upper bound uses the Unicode sentinel \uf8ff (the highest character in the Basic Multilingual Plane), so any string that starts with the prefix sorts before it.

def build_prefix_filters(field: str, prefix: str) -> FilterDict:
    ...

fastapi.pagination — FastAPI Adapter

PaginationParams is a FastAPI dependency that extracts cursor, direction, and limit from query-string parameters, ready to pass straight to cursor_paginate.

Quick Start

from fastapi import Depends, FastAPI
from firedantic_extras import cursor_paginate, CursorPage
from firedantic_extras.fastapi.pagination import PaginationParams

app = FastAPI()

class Product(Model):
    __collection__ = "products"
    name: str
    price: float
    category: str

@app.get("/products")
def list_products(
    pagination: PaginationParams = Depends(),
    category: str | None = None,
) -> CursorPage[Product]:
    filter_ = {"category": category} if category else None
    return cursor_paginate(
        Product,
        limit=pagination.limit,
        order_by="name",
        cursor=pagination.cursor,
        direction=pagination.direction,
        filter_=filter_,
    )

cursor_paginate blocks, so the route above is a plain def — FastAPI runs it in a threadpool. For an async def route, use an AsyncModel and async_cursor_paginate:

from firedantic import AsyncModel
from firedantic_extras.fastapi.pagination import PaginationParams, async_cursor_paginate

class Product(AsyncModel):
    __collection__ = "products"
    name: str
    price: float
    category: str

@app.get("/products")
async def list_products(pagination: PaginationParams = Depends()) -> CursorPage[Product]:
    return await async_cursor_paginate(
        Product,
        limit=pagination.limit,
        order_by="name",
        cursor=pagination.cursor,
        direction=pagination.direction,
    )

Request examples:

GET /products?limit=20
GET /products?limit=20&cursor=<next_cursor>&direction=next
GET /products?limit=20&cursor=<prev_cursor>&direction=prev

Response:

{
  "items": [ ... ],
  "has_next": true,
  "has_prev": false,
  "next_cursor": "abc123",
  "prev_cursor": null,
  "total": null
}

API

class PaginationParams:
    def __init__(
        self,
        cursor: str | None = Query(default=None),
        direction: Literal["next", "prev"] = Query(default="next"),
        limit: int = Query(default=20, ge=1, le=500),
    ) -> None: ...

flask.pagination — Flask Adapter

FlaskPaginationParams extracts cursor, direction, limit, order_by, and order_dir from request.args. paginate_model() runs cursor_paginate and wraps the result in a PaginatedContext with everything a Jinja template needs for sortable columns and Previous/Next links.

Quick Start

from flask import Blueprint, render_template
from firedantic_extras.flask.pagination import FlaskPaginationParams, paginate_model

blueprint = Blueprint("products", __name__)

class Product(Model):
    __collection__ = "products"
    name: str
    price: float
    category: str

@blueprint.route("/products")
def list_products():
    params = FlaskPaginationParams.from_request(default_order_by="name")
    ctx = paginate_model(Product, params)
    return render_template("products.html", ctx=ctx)

In the template:

<a href="{{ url_for('products.list_products', **ctx.sort_params('name')) }}">
  Name {{ ctx.sort_indicator('name') }}
</a>

{% if ctx.page.has_prev %}
  <a href="{{ url_for('products.list_products', **ctx.prev_params()) }}">Previous</a>
{% endif %}
{% if ctx.page.has_next %}
  <a href="{{ url_for('products.list_products', **ctx.next_params()) }}">Next</a>
{% endif %}

Request examples:

GET /products?limit=20
GET /products?order_by=name&order_dir=ASCENDING
GET /products?cursor=<next_cursor>&direction=next

API

@dataclass
class FlaskPaginationParams:
    cursor: str | None = None
    direction: Literal["next", "prev"] = "next"
    limit: int = 50
    order_by: str = "updated_at"
    order_dir: str = "DESCENDING"

    @classmethod
    def from_request(cls, default_order_by="updated_at", default_order_dir="DESCENDING") -> FlaskPaginationParams: ...
    def build_query_params(self, **overrides) -> dict[str, str]: ...


@dataclass
class PaginatedContext:
    page: CursorPage
    params: FlaskPaginationParams
    total: int | None = None

    @property
    def items(self) -> list[Any]: ...  # shortcut for page.items
    @property
    def showing_count(self) -> int: ...  # len(page.items)
    @property
    def used_fallback(self) -> bool: ...  # shortcut for page.used_fallback
    def next_params(self) -> dict[str, str]: ...
    def prev_params(self) -> dict[str, str]: ...
    def sort_params(self, field_name: str) -> dict[str, str]: ...
    def sort_indicator(self, field_name: str) -> str: ...
    def limit_params(self, new_limit: int) -> dict[str, str]: ...


def paginate_model(
    model_class: type[BareModel],
    params: FlaskPaginationParams,
    *,
    filter_: FilterDict | None = None,
    order_by: OrderByInput | None = None,
    include_total: bool = True,
    exclude_null_sort_field: bool = False,
    fallback_order_by: OrderByInput | None = None,
) -> PaginatedContext: ...

Jinja Macros for Sortable, Paginated Tables

A bundled firedantic_extras/macros.html template ships four Bootstrap-5 macros that read directly from a PaginatedContext — sortable column headers, Previous/Next controls, a result-count line, and a page-size selector. Register the template loader once during app setup:

from firedantic_extras.flask.templates import register_macros

app = Flask(__name__)
register_macros(app)

Then in any template:

{% import "firedantic_extras/macros.html" as fe_macros %}

{{ fe_macros.result_info(ctx) }}

<table class="table table-hover">
  <thead>
    <tr>
      {{ fe_macros.sortable_header(ctx, "name", "Name", "products.list_products") }}
      {{ fe_macros.sortable_header(ctx, "category", "Category", "products.list_products") }}
    </tr>
  </thead>
  <tbody>
    {% for item in ctx.items %}
    <tr>
      <td>{{ item.name }}</td>
      <td>{{ item.category }}</td>
    </tr>
    {% endfor %}
  </tbody>
</table>

{{ fe_macros.pagination_controls(ctx, "products.list_products") }}
{{ fe_macros.page_size_selector(ctx, "products.list_products") }}

These macros are intentionally minimal — copy and adapt them for custom styling or a different CSS framework.

Installation

Published on PyPI as altissimo-firedantic-extras; the import name is firedantic_extras. It depends on altissimo-firedantic, the Altissimo fork of firedantic (imported as firedantic). Don't install it alongside upstream firedantic — both provide the firedantic package.

# Core (includes update_collection)
pip install altissimo-firedantic-extras

# With FastAPI pagination support
pip install altissimo-firedantic-extras[fastapi]

# With Flask pagination support
pip install altissimo-firedantic-extras[flask]

# With BigQuery schema generation
pip install altissimo-firedantic-extras[bigquery]

# Everything
pip install altissimo-firedantic-extras[all]

update_collection — Collection Sync

Synchronize a Firestore collection to match a given list of Firedantic models. Documents are added, updated, or deleted as needed, using batched writes that respect Firestore's 500-document batch limit.

Why?

Manually diffing existing documents against a desired state is tedious and error-prone. CollectionSync handles the full add/update/delete lifecycle in a single class, with support for dry-run mode, field-level diffing, and configurable sync keys.

The comparison is done against raw Firestore data (not re-hydrated models), so stale or extra fields stored in Firestore are visible and will trigger updates — ensuring Firestore always converges to the exact shape described by the model.

The incoming side is compared in the form firedantic stores it (to_firestore_value(): enums as their value, dates as ISO strings, decimals as strings, …), so a re-sync of unchanged models reports skips rather than rewriting every document with such fields. Writes go through firedantic's batched save(batch=...) / delete(batch=...), so ID generation, field aliases and __db_config__ routing behave exactly as they do for a single save().

Quick Start

from firedantic import Model
from firedantic_extras import CollectionSync

class User(Model):
    __collection__ = "users"
    name: str
    email: str
    active: bool = True

desired = [
    User(id="u1", name="Alice", email="alice@example.com"),
    User(id="u2", name="Bob",   email="bob@example.com"),
    User(id="u3", name="Carol", email="carol@example.com"),
]

# Additive sync (default) — adds new docs, updates changed docs,
# but does NOT delete docs missing from the list.
result = CollectionSync.sync(User, desired)
print(result.summary())
# SyncResult(adds=3, updates=0, deletes=0, skips=0)

# Full sync — also deletes docs not in the desired list.
result = CollectionSync.sync(User, desired, delete_items=True)

# Dry run with field-level diff output — preview changes without writing.
result = CollectionSync.sync(
    User, desired, delete_items=True, diff=True, dry_run=True,
)
print(result.summary())
# SyncResult(adds=0, updates=1, deletes=2, skips=0, DRY RUN)

# Inspect field-level diffs for updated documents.
for key, doc_diff in result.diffs.items():
    for change in doc_diff.changes:
        print(f"  {change.field}: {change.before!r} → {change.after!r}")

With firedantic.AsyncModel subclasses use AsyncCollectionSync — same constructor and options, run() and sync() are coroutines:

from firedantic_extras import AsyncCollectionSync

result = await AsyncCollectionSync.sync(User, desired, delete_items=True, diff=True)

Preserving fields another writer owns

Sometimes a collection mirrors an external source but also carries a few fields your own code writes, e.g. a webhook recording when a notification email went out. A plain sync replaces each changed document, so it would wipe those fields, and reading them first to copy them across still loses a write that lands between the sync's read and its commit.

Name those fields in preserve_fields:

result = CollectionSync.sync(
    Order,
    orders_from_api,                 # email_sent_at is None on all of these
    delete_items=True,
    preserve_fields=["email_sent_at", "email_skipped_reason"],
)

On existing documents, preserved fields are dropped from both sides of the comparison: a difference in them alone is a skip, they never count as stale, and they don't appear in diffs. A document that does need updating gets a field-level update() with only its changed fields (stale ones removed with DELETE_FIELD) instead of a full set, so whatever is stored in the preserved fields, including a write made while the sync runs, is kept.

• Model field names resolve to their aliases. Any other name is taken as a stored key, so fields the model doesn't declare can be preserved too.
• Only top-level fields; a name with dots is one field, not a nested path.
• New documents are written whole, with the model's values for the preserved fields.
• delete_items=True still deletes whole documents, preserved fields included.
• A field-level update() fails if the document is deleted after the sync read it, which fails that batch commit; a plain sync would recreate it.

API Reference

CollectionSync / AsyncCollectionSync

class CollectionSync:
    """Reconcile a Firestore collection to match a desired list of models."""

    def __init__(
        self,
        model: type[BareModel],
        items: Sequence[BareModel],
        *,
        delete_items: bool = False,
        dry_run: bool = False,
        diff: bool = False,
        output_writer: Callable[[str], None] | None = print,
        sync_key: str | None = None,
        on_duplicate_keys: OnDuplicateKeys = "raise",
        on_error: OnError = "raise",
        chunk_size: int = 500,
        preserve_fields: Collection[str] = (),
    ) -> None: ...

    def run(self) -> SyncResult:
        """Execute the sync and return a SyncResult."""

    @classmethod
    def sync(
        cls,
        model: type[BareModel],
        items: Sequence[BareModel],
        **kwargs,
    ) -> SyncResult:
        """Convenience class method — construct and run in one call."""

Note: UpdateCollection is available as a backward-compatible alias for CollectionSync.

Parameters

SyncResult

@dataclass
class SyncResult:
    adds: int = 0
    updates: int = 0
    deletes: int = 0
    skips: int = 0
    diffs: dict[str, DocumentDiff]   # populated when diff=True
    errors: list[SyncError]          # populated when on_error != "raise"
    dry_run: bool = False
    skipped_duplicate_keys: list[str]  # populated when on_duplicate_keys="skip"

    @property
    def has_errors(self) -> bool: ...

    @property
    def total_changes(self) -> int:
        """adds + updates + deletes (excludes skips)."""

    def summary(self) -> str:
        """One-liner: 'SyncResult(adds=1, updates=2, ...)'"""

Supporting types

@dataclass
class FieldDiff:
    """A single field-level change."""
    field: str
    before: Any   # value in Firestore (_MISSING if absent)
    after: Any    # value in desired model (_MISSING if absent)

@dataclass
class DocumentDiff:
    """All field-level changes for one document."""
    doc_id: str
    sync_key_value: str
    changes: list[FieldDiff]

@dataclass
class SyncError:
    """An error for one document (when on_error != 'raise')."""
    sync_key_value: str
    error: Exception

build_sync_plan (advanced)

The pure-function core of CollectionSync, exposed for testing and inspection without any Firestore I/O:

def build_sync_plan(
    desired: dict[str, BareModel],
    existing_models: dict[str, BareModel],
    existing_raw: dict[str, dict[str, Any]],
    doc_id_field: str,
    delete_items: bool = False,
    diff: bool = False,
    preserve_keys: Collection[str] = frozenset(),
) -> _SyncPlan:
    """Compute adds/updates/deletes/skips from pure data — no Firestore calls."""

bigquery.schema — Schema Generation

Automatically generates BigQuery table schemas from Firedantic model classes, mapping Pydantic field types to their BigQuery equivalents.

Why?

When you maintain Firedantic models as your source of truth for Firestore documents and also need to export that data to BigQuery, keeping schemas in sync manually is fragile. This module derives the BigQuery schema directly from your model definitions — including REQUIRED/NULLABLE mode based on whether Pydantic fields are required or optional, and REPEATED mode for list fields.

Quick Start

from firedantic_extras.bigquery import model_to_bq_schema, schema_to_dict

class Sample(Model):
    __collection__ = "samples"
    sample_id: str          # required → REQUIRED
    barcode: str
    collected_at: datetime
    results: dict[str, float]   # dict → JSON
    tags: list[str]             # list[str] → REPEATED STRING
    notes: str | None = None    # optional → NULLABLE

schema = model_to_bq_schema(Sample)
# [
#   SchemaField("id",           "STRING",    mode="NULLABLE"),   ← always first
#   SchemaField("sample_id",    "STRING",    mode="REQUIRED"),
#   SchemaField("barcode",      "STRING",    mode="REQUIRED"),
#   SchemaField("collected_at", "TIMESTAMP", mode="REQUIRED"),
#   SchemaField("results",      "JSON",      mode="NULLABLE"),
#   SchemaField("tags",         "STRING",    mode="REPEATED"),
#   SchemaField("notes",        "STRING",    mode="NULLABLE"),
# ]

Type Mapping

The schema describes documents as firedantic stores them, not the Python annotations. firedantic (0.20+) writes a model with model_dump(by_alias=True) and converts values Firestore can't hold natively, so column names are field aliases and column types follow the stored value:

NUMERIC holds up to 9 decimal places; use json_fields (or a BIGNUMERIC column via extra_fields) for decimals with more.

Mode rules:

• Required Pydantic field (field: T) → REQUIRED
• Optional[T] / T | None / field with a default → NULLABLE
• list[T] → REPEATED (BQ does not support REQUIRED for repeated fields)
• id is always STRING NULLABLE (first field, regardless of model definition)
• Fields with an alias get the alias as column name; json_fields and exclude_fields accept either the field name or the alias

Backward Compatibility — json_fields

When migrating from hand-written schemas where nested objects were stored as JSON, use json_fields to keep specific fields as JSON NULLABLE regardless of what the model says:

# populations: list[Population] would normally → REPEATED RECORD
# but our existing BQ table has it as JSON — keep it for now
schema = model_to_bq_schema(Kit, json_fields={"populations", "acquired_from"})

This lets you migrate one table at a time without breaking existing queries.

Full API

def model_to_bq_schema(
    model_class: type[BaseModel],
    *,
    json_fields: set[str] | None = None,
    exclude_fields: set[str] | None = None,
    extra_fields: list[SchemaField] | None = None,
) -> list[SchemaField]:
    """Generate a BigQuery schema from a Firedantic / Pydantic model.

    Args:
        model_class: The Pydantic model class to introspect.
        json_fields: Field names to force to JSON NULLABLE (backward-compat).
        exclude_fields: Field names to omit from the schema entirely.
        extra_fields: Additional SchemaFields to append at the end
            (e.g., load-time metadata columns not in the model).
    """


def models_to_bq_schemas(
    model_classes: list[type[BaseModel]],
    **kwargs,
) -> dict[str, list[SchemaField]]:
    """Generate schemas for multiple models, keyed by __collection__ name.

    Args:
        model_classes: Firedantic model classes (must have __collection__).
        **kwargs: Forwarded to model_to_bq_schema.

    Returns:
        Dict mapping __collection__ names to their BigQuery schemas.
    """


def schema_to_dict(schema: list[SchemaField]) -> list[dict]:
    """Serialise a schema to a JSON-serialisable list of dicts.

    Output matches the BigQuery REST API representation and can be stored
    in a JSON file or round-tripped via Client.schema_from_json().
    """


def compare_schemas(
    a: list[SchemaField],
    b: list[SchemaField],
) -> SchemaDiff:
    """Diff two BigQuery schemas at the top level (field names and BQ types).

    Useful for verifying a model-derived schema against an existing live BQ
    table schema before cutting over from hand-written definitions.

    Returns a SchemaDiff with:
      .only_in_a       — fields in a but not b
      .only_in_b       — fields in b but not a
      .type_mismatches — [(field, type_in_a, type_in_b), ...]
      .is_equal        — True if schemas are identical
    """

Migration Example for json2bq

from firedantic_extras.bigquery.schema import model_to_bq_schema, compare_schemas

# Map (dataset, table) to (ModelClass, fields_to_keep_as_json)
MODEL_MAP = {
    ("darwinsark", "kits"):    (Kit,    {"populations", "acquired_from"}),
    ("darwinsark", "animals"): (Animal, {"consent", "breeds"}),
    # tables without a model fall back to BQ autodetect (old behaviour)
}

def create_schema(dataset_name, table_name):
    entry = MODEL_MAP.get((dataset_name, table_name))
    if entry is None:
        return None
    model_class, json_fields = entry
    return model_to_bq_schema(model_class, json_fields=json_fields)

# Verify new schema matches existing table before switching over:
existing = client.get_table("darwinsark.kits").schema
generated = create_schema("darwinsark", "kits")
diff = compare_schemas(existing, generated)
if not diff.is_equal:
    print("Fields only in live table:", diff.only_in_a)
    print("Fields only in model:     ", diff.only_in_b)
    print("Type mismatches:          ", diff.type_mismatches)

Development

# Clone and install
git clone https://github.com/altissimo-hq/firedantic-extras.git
cd firedantic-extras
poetry install --with dev --all-extras

# Run unit tests (default — no emulator needed)
poetry run pytest

# Lint and format
poetry run ruff check --fix .
poetry run ruff format .

# Pre-commit hooks (installed automatically)
poetry run pre-commit run --all-files

Generated Sync Code (unasync.py)

The async code is the source of truth; the sync flavour is generated from it, the same way Firedantic itself does it:

unasync.py rewrites each file line by line (async def → def, await x → x, AsyncFoo → Foo, async_cursor_paginate → cursor_paginate, …) and ruff-formats the result. The pre-commit hook runs it on every commit, and because the generated files are committed, CI's pre-commit run --all-files fails if they are stale.

poetry run python unasync.py   # regenerate by hand

Keep the _async modules thin: anything that doesn't need a Firestore round-trip (validation, planning, diffing, result assembly) belongs in src/firedantic_extras/common/, which both flavours import. That keeps the generated code small and the substitution table short. The top-level modules (cursor_pagination.py, query.py, update_collection.py) are facades that re-export the public API from common/, _async/ and _sync/.

Integration Tests (Firestore Emulator)

Integration tests exercise the full Firestore round-trip and require the Firebase Emulator Suite.

# Prerequisites: Firebase CLI (https://firebase.google.com/docs/cli)
npm install -g firebase-tools

# Terminal 1 — start the emulator (Firestore on port 8686)
./scripts/start_emulator.sh

# Terminal 2 — run integration tests only
FIRESTORE_EMULATOR_HOST=127.0.0.1:8686 poetry run pytest -m integration -v

# Or run everything (unit + integration)
FIRESTORE_EMULATOR_HOST=127.0.0.1:8686 poetry run pytest -m "" -v

The default pytest command excludes integration tests via addopts in pyproject.toml, so poetry run pytest (and pre-commit) always runs fast, emulator-free unit tests.

License

BSD 3-Clause. See LICENSE for details.