Get Started with Firestore Pydantic ODM

This guide walks you through everything you need to go from a blank Python environment to a fully working Firestore-backed application using Firestore Pydantic ODM. Each step builds on the previous one, and a complete runnable snippet is provided at the end so you can verify your setup immediately.

Install the package

Install Firestore Pydantic ODM from PyPI using pip:

pip install firestore-pydantic-odm

If you plan to develop locally against the Firestore emulator, install the optional emulator extra as well:

pip install "firestore-pydantic-odm[emulator]"

Set up Google credentials

The library uses the official google-cloud-firestore async client, which follows the standard Google Application Default Credentials (ADC) chain.

For production or staging, point the environment variable at your service-account key file:

export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account.json"

For local development with the emulator, start the emulator and export its host:

export FIRESTORE_EMULATOR_HOST="localhost:8080"

You can also pass emulator_host= directly to FirestoreDB at runtime instead of setting the environment variable—the library will configure the environment variable for you automatically.

If you are running inside Google Cloud (Cloud Run, GKE, Cloud Functions, etc.) no extra configuration is needed—the metadata server provides credentials automatically.

Define a model

Subclass BaseFirestoreModel and declare the Firestore collection name in an inner Settings class. Add your document fields as ordinary Pydantic field annotations:

from firestore_pydantic_odm import BaseFirestoreModel

class User(BaseFirestoreModel):
    class Settings:
        name = "users"  # Firestore collection name

    name: str
    email: str
    age: int = 0

Every model automatically gets an id: Optional[str] field that stores the Firestore document ID. You do not need to declare it yourself.

The Settings.name attribute is required. If you omit it, the library falls back to the class name, which is usually not what you want for production collections.

Initialize the database

Create a FirestoreDB instance and call init_firestore_odm() with it and a list of every model class your application uses. This step injects the database client into each model and registers the model graph for cascade-delete resolution:

from firestore_pydantic_odm import FirestoreDB, init_firestore_odm

# Connect to the real Firestore backend
db = FirestoreDB(project_id="my-gcp-project")

# Or connect to the local emulator
db = FirestoreDB(project_id="my-gcp-project", emulator_host="localhost:8080")

# Register all models — pass every model class you intend to use
init_firestore_odm(db, [User])

Always call init_firestore_odm() before performing any database operation. Attempting to call save(), find(), or any other async method before initialization raises a RuntimeError.

Perform async CRUD

All document operations are async and must be awaited. Use save() to create, update() to persist changes, delete() to remove, and get() to fetch a document by its ID:

import asyncio
from firestore_pydantic_odm import FirestoreDB, init_firestore_odm, BaseFirestoreModel

class User(BaseFirestoreModel):
    class Settings:
        name = "users"

    name: str
    email: str

async def crud_example():
    db = FirestoreDB(project_id="my-gcp-project", emulator_host="localhost:8080")
    init_firestore_odm(db, [User])

    # CREATE — Firestore auto-generates the document ID
    user = User(name="Alice", email="[email protected]")
    await user.save()
    print(f"Created user with id: {user.id}")

    # UPDATE — mutate fields and call update()
    user.email = "[email protected]"
    await user.update()

    # GET — retrieve a document by its Firestore document ID
    fetched = await User.get(user.id)
    print(f"Fetched: {fetched.name} — {fetched.email}")

    # DELETE — permanently removes the document
    await user.delete()

asyncio.run(crud_example())

Query with find() and find_one()

find() is an async generator that streams matching documents. find_one() returns the first match or None. Both accept filters, order_by, limit, offset, and an optional projection:

import asyncio
from pydantic import BaseModel
from firestore_pydantic_odm import (
    FirestoreDB, init_firestore_odm, BaseFirestoreModel,
    OrderByDirection,
)

class User(BaseFirestoreModel):
    class Settings:
        name = "users"

    name: str
    email: str
    age: int = 0

# Projection — only fetch the fields you need
class UserProjection(BaseModel):
    name: str

async def query_example():
    db = FirestoreDB(project_id="my-gcp-project", emulator_host="localhost:8080")
    init_firestore_odm(db, [User])

    # Stream all users ordered by name ascending
    async for u in User.find(order_by=[(User.name, OrderByDirection.ASCENDING)]):
        print(u)

    # Filter — find users aged 18 or older
    async for u in User.find(filters=[User.age >= 18]):
        print(u.name)

    # Projection — Firestore only transfers the `name` field
    async for u in User.find(
        filters=[User.age >= 18],
        projection=UserProjection,
        order_by=[(User.name, OrderByDirection.ASCENDING)],
    ):
        print(u.name)  # u is a UserProjection instance

    # find_one — return the first matching document
    alice = await User.find_one(filters=[User.email == "[email protected]"])
    if alice:
        print(f"Found: {alice.name}")

asyncio.run(query_example())

Complete Working Example

The snippet below combines every step above into a single asyncio.run() call you can copy, adjust the project_id and emulator_host, and run immediately:

import asyncio
from pydantic import BaseModel
from firestore_pydantic_odm import (
    FirestoreDB,
    BaseFirestoreModel,
    BatchOperation,
    OrderByDirection,
    init_firestore_odm,
)


class User(BaseFirestoreModel):
    class Settings:
        name = "users"

    name: str
    email: str
    age: int = 0


class UserProjection(BaseModel):
    name: str


async def main():
    # 1. Connect (emulator)
    db = FirestoreDB(project_id="my-gcp-project", emulator_host="localhost:8080")
    init_firestore_odm(db, [User])

    # 2. Create documents
    alice = User(name="Alice", email="[email protected]", age=30)
    await alice.save()

    bob = User(name="Bob", email="[email protected]", age=17)
    await bob.save()

    # 3. Update a field
    alice.email = "[email protected]"
    await alice.update()

    # 4. Get by ID
    fetched = await User.get(alice.id)
    print(f"Fetched: {fetched.name} — {fetched.email}")

    # 5. Query — all users ordered by name
    print("All users:")
    async for u in User.find(order_by=[(User.name, OrderByDirection.ASCENDING)]):
        print(f"  {u.name}")

    # 6. Query with projection (only `name` field transferred)
    print("Adults (projection):")
    async for u in User.find(
        filters=[User.age >= 18],
        projection=UserProjection,
    ):
        print(f"  {u.name}")

    # 7. find_one
    first = await User.find_one(
        filters=[],
        order_by=[(User.name, OrderByDirection.ASCENDING)],
    )
    print(f"First user alphabetically: {first.name}")

    # 8. Batch write
    ops = [
        (BatchOperation.CREATE, User(name="Carol", email="[email protected]", age=25)),
        (BatchOperation.DELETE, bob),
    ]
    await User.batch_write(ops)

    # 9. Delete
    await alice.delete()


asyncio.run(main())

Next Steps

Installation

Detailed install options, dependency versions, and credential setup for all environments.

Models

Learn how to define fields, aliases, nested objects, and model configuration.

Querying

Filters, ordering, pagination, projections, and collection-group queries.

Subcollections

Declare parent–child relationships and query deeply nested documents.

Batch Operations

Atomic multi-document writes and transactions.

Testing

Emulator setup and unit-test mocking strategies.

​Complete Working Example

​Next Steps