firestore-pydantic-odm 0.2.2

Asynchronous Pydantic ODM for Google Cloud Firestore

Firestore Pydantic ODM

Firestore Pydantic ODM is a lightweight, fully-typed Object-Document Mapper for Google Cloud Firestore.
It combines [Pydantic]’s data-validation super-powers with Firestore’s scalable NoSQL store, offering async CRUD, batch writes, transactions, and projections that request only the fields you need—making queries faster and cheaper.

---

Features

• Asynchronous CRUD: Full support for creating, reading, updating, and deleting Firestore documents using async/await.
• Validation with Pydantic: Define your data models with automatic validation, ensuring data integrity before it reaches the database.
• Advanced Queries: Perform searches with filters, projections (selecting only specific fields), and ordering.
• Batch Operations and Transactions: Group multiple write operations and execute transactions atomically for greater efficiency and consistency.
• Emulator and Testing Support: Easily switch to the Firestore emulator or plug in mocks for unit testing.
• Seamless Integration: Fits smoothly into any Python project with minimal setup.

---

Installation

pip install firestore-pydantic-odm

---

Quick Start

1 · Define a model

from firestore_pydantic_odm import BaseFirestoreModel

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

    name: str
    email: str

2 · Initialise Firestore

from firestore_pydantic_odm import FirestoreDB, BaseFirestoreModel

db = FirestoreDB(project_id="my-project", emulator_host="localhost:8080")  # optional emulator
BaseFirestoreModel.initialize_db(db)

3 · Async CRUD

user = User(name="Alice", email="[e-mail]")
await user.save()               # CREATE

user.email = "[e-mail]"
await user.update()             # UPDATE

await user.delete()             # DELETE

4 · Querying & Projections

# Simple filter
async for u in User.find(filters=[User.name == "Alice"]):
    print(u)

# Single document
u = await User.find_one(filters=[User.email == "[e-mail]"])

Projections — selecting only the fields you need

from pydantic import BaseModel

class UserProjection(BaseModel):
    name: str            # only grab the `name` field

async for u in User.find(
        filters=[User.age >= 18],
        projection=UserProjection):
    print(u.name)        # `u` is an instance of UserProjection

# Fetch a single document with a projection
u = await User.find_one(
        filters=[User.id == "abc123"],
        projection=UserProjection)

How it works: the ODM converts UserProjection into a Firestore field mask, so the RPC fetches only the columns defined in that class. Each item yielded by find() (or returned by find_one()) is therefore of type UserProjection, giving you a clean List[UserProjection] with exactly the data requested.

5 · Batch writes

from firestore_pydantic_odm import BatchOperation

ops = [
    (BatchOperation.CREATE, User(name="Bob", email="[e-mail]")),
    (BatchOperation.UPDATE, user),            # previously fetched instance
    (BatchOperation.DELETE, another_user)     # instance with `id` set
]
await User.batch_write(ops)

---

Testing

The project ships with pytest and pytest-asyncio fixtures. To run the suite:

pytest

Set FIRESTORE_EMULATOR_HOST=localhost:8080 to run tests against the local emulator instead of production Firestore.

---

Contributing

1. Fork the repository
2. git checkout -b feature/awesome
3. Write code & tests; ensure all tests pass
4. Open a Pull Request describing your improvements

---

License

Distributed under the BSD 3-Clause License. See the LICENSE file for full text.