Get Started with Firestore Pydantic ODM

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())