BaseFirestoreModel — Async Firestore ODM Base Class

BaseFirestoreModel is the foundation of the Firestore Pydantic ODM. Every document model in your application inherits from it, gaining a complete async CRUD interface — including filtering, pagination, batch operations, and subcollection traversal — while remaining a fully valid Pydantic model for serialization and validation.

Class overview

BaseFirestoreModel extends Pydantic BaseModel and adds Firestore-aware behaviour through class-level injection of a FirestoreDB instance. All database operations are async and must be awaited (or iterated with async for in the case of generators). The class is compatible with both Pydantic v1 and v2.

from firestore_pydantic_odm import BaseFirestoreModel, init_firestore_odm, FirestoreDB

db = FirestoreDB(project_id="my-project")

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

    username: str
    email: str

init_firestore_odm(database=db, document_models=[User])

The `Settings` inner class

Each model should define an inner Settings class to control how it maps to Firestore.

Attribute	Type	Description
`name`	`str`	The Firestore collection name. Defaults to `"BaseCollection"` on the base class — always override this in subclasses.
`parent`	`Type[BaseFirestoreModel]`	Set on subcollection models to declare the parent document type. Omit for top-level collections.

class Post(BaseFirestoreModel):
    class Settings:
        name = "posts"
        parent = User   # makes Post a subcollection of User

    title: str
    published: bool = False

The `id` field

id: Optional[str] = Field(default=None)

Firestore document IDs are stored in the id field. When id is None at save time, Firestore auto-generates one. The field is excluded from serialised document data — it is stored only as the document key.

Class methods

`initialize_db`

Inject the FirestoreDB instance that all operations on this class will use. Called automatically by init_firestore_odm — you rarely need to call it directly.

@classmethod
def initialize_db(cls, db: FirestoreDB) -> None

FirestoreDB

required

The FirestoreDB wrapper instance containing the underlying AsyncClient.

`initialize_fields`

Attach FirestoreField descriptors to every Pydantic field on the class so that filter expressions like User.email == "[email protected]" produce the correct Firestore FieldFilter tuples. Called automatically by init_firestore_odm.

@classmethod
def initialize_fields(cls) -> None

After initialize_fields runs, accessing a class attribute such as User.email returns a FirestoreField object that supports ==, !=, <, <=, >, >=, in, and array_contains operators to build filter tuples.

`get`

Retrieve a single document by its Firestore document ID.

@classmethod
async def get(
    cls,
    doc_id: str,
    parent: Optional[BaseFirestoreModel] = None,
) -> Optional[Self]

str

required

The Firestore document ID to look up.

BaseFirestoreModel | None

Parent document instance required when this model is a subcollection. Omit for top-level collections.

Returns the hydrated model instance, or None if no document with that ID exists.

user = await User.get("uid_abc123")

# Subcollection variant
post = await Post.get("pid_xyz", parent=user)

`exists`

Return True if a document with the given ID exists in Firestore, without fetching its data.

@classmethod
async def exists(
    cls,
    doc_id: str,
    parent: Optional[BaseFirestoreModel] = None,
) -> bool

str

required

The Firestore document ID to check.

BaseFirestoreModel | None

Parent document instance for subcollection models.

if await User.exists("uid_abc123"):
    print("User exists")

`find`

Asynchronously stream documents matching the given filters. Returns an AsyncGenerator — iterate with async for.

@classmethod
async def find(
    cls,
    filters: List[Tuple[FieldType, FirestoreOperators, Any]] = None,
    parent: Optional[BaseFirestoreModel] = None,
    projection: Optional[Type[BaseModel]] = None,
    order_by: Optional[Union[List[Union[FieldType, FieldOrderType]], FieldType, FieldOrderType]] = None,
    limit: Optional[int] = None,
    offset: Optional[int] = None,
) -> AsyncGenerator

List[Tuple[FieldType, FirestoreOperators, Any]] | None

A list of filter tuples. Each tuple is (field, operator, value). Build tuples with FirestoreField expressions such as User.email == "[email protected]" or raw strings ("email", "==", "[email protected]"). Defaults to [] (no filters — returns all documents).

BaseFirestoreModel | None

Parent document instance for subcollection models.

Type[BaseModel] | None

A Pydantic model class containing only the fields you want retrieved. Reduces bandwidth by using Firestore field projection.

FieldType | FieldOrderType | List | None

A field or list of fields to sort by. Pass a (field, OrderByDirection) tuple for explicit direction, or just a field for the default ascending order.

int | None

Maximum number of documents to return.

int | None

Number of documents to skip before returning results.

from firestore_pydantic_odm import OrderByDirection

# All active users, newest first, page 2
async for user in User.find(
    filters=[User.active == True],
    order_by=(User.created_at, OrderByDirection.DESCENDING),
    limit=20,
    offset=20,
):
    print(user.username)

`find_one`

Return the first document matching the given filters, or None if no match is found. Internally calls find(..., limit=1).

@classmethod
async def find_one(
    cls,
    filters: List[Tuple[str, str, Any]],
    parent: Optional[BaseFirestoreModel] = None,
    projection: Optional[Type[BaseModel]] = None,
    order_by: Optional[Union[FieldType, FieldOrderType]] = None,
) -> Optional[Self]

List[Tuple[str, str, Any]]

required

Filter tuples identical to those accepted by find.

BaseFirestoreModel | None

Parent document instance for subcollection models.

Type[BaseModel] | None

Optional Pydantic model class for field projection.

FieldType | FieldOrderType | None

Field to sort by to determine which document is “first”.

user = await User.find_one([User.email == "[email protected]"])
if user:
    print(user.username)

`count`

Return the number of documents that match the given filters using Firestore’s native count() aggregation. Falls back to fetching all document IDs if the SDK version does not support count().

@classmethod
async def count(
    cls,
    filters: List[Tuple[str, str, Any]],
    parent: Optional[BaseFirestoreModel] = None,
) -> int

List[Tuple[str, str, Any]]

required

Filter tuples. Pass an empty list [] to count all documents in the collection.

BaseFirestoreModel | None

Parent document instance for subcollection models.

active_count = await User.count([User.active == True])
print(f"{active_count} active users")

`batch_write`

Execute multiple create, update, and delete operations atomically in a single Firestore batch commit.

@classmethod
async def batch_write(
    cls,
    operations: List[Tuple[BatchOperation, BaseFirestoreModel]],
) -> None

List[Tuple[BatchOperation, BaseFirestoreModel]]

required

A list of (BatchOperation, model_instance) tuples. BatchOperation is an enum with values CREATE, UPDATE, and DELETE. The model instances do not need to belong to the same collection — the collection is inferred from each instance’s class.

For UPDATE and DELETE operations the model instance must have a non-None id. Only CREATE can operate on instances without a pre-assigned ID; Firestore will generate one automatically.

from firestore_pydantic_odm import BatchOperation

alice = User(username="alice", email="[email protected]")
old_user = await User.get("uid_old")

await User.batch_write([
    (BatchOperation.CREATE, alice),
    (BatchOperation.DELETE, old_user),
])

`collection_group_find`

Query across all subcollections sharing this model’s collection name, regardless of which parent document they belong to. Uses Firestore’s collection_group() API.

@classmethod
async def collection_group_find(
    cls,
    filters: List[Tuple[FieldType, FirestoreOperators, Any]] = None,
    projection: Optional[Type[BaseModel]] = None,
    order_by: Optional[Union[List[Union[FieldType, FieldOrderType]], FieldType, FieldOrderType]] = None,
    limit: Optional[int] = None,
    offset: Optional[int] = None,
) -> AsyncGenerator

List[Tuple[FieldType, FirestoreOperators, Any]] | None

Filter tuples applied across all matching subcollections. Defaults to [].

Type[BaseModel] | None

Optional Pydantic model class for field projection.

FieldType | FieldOrderType | List | None

Field(s) to sort results by.

int | None

Maximum number of documents to return across all parents.

int | None

Number of matching documents to skip.

Collection group queries require a Firestore composite index. The _parent_path attribute is automatically populated on each returned instance from the document’s reference path.

# Find all published posts across every user
async for post in Post.collection_group_find([Post.published == True]):
    print(post.title, post._parent_path)

Instance methods

`save`

Persist a new document to Firestore. If id is None, Firestore generates a unique document ID and assigns it to the instance. Raises RuntimeError if the database has not been initialised, or if an explicit id is provided but a document with that ID already exists.

async def save(
    self,
    parent: Optional[BaseFirestoreModel] = None,
    exclude_none: bool = True,
    by_alias: bool = True,
    exclude_unset: bool = True,
) -> Self

BaseFirestoreModel | None

Parent document instance. Required for subcollection models if _parent_path is not already stored on the instance.

bool

default:"True"

When True, fields with None values are not written to Firestore.

bool

default:"True"

When True, field aliases are used as Firestore field names in the document.

bool

default:"True"

When True, fields that were not explicitly set during model construction are omitted from the write.

Returns the same instance with id populated.

user = User(username="alice", email="[email protected]")
await user.save()
print(user.id)  # auto-generated Firestore ID

`update`

Update fields on an existing Firestore document. The instance must have a non-None id — raises ValueError if id is not set. Only fields that pass the exclusion rules are sent to Firestore.

async def update(
    self,
    parent: Optional[BaseFirestoreModel] = None,
    include: Optional[set] = None,
    exclude_none: bool = True,
    by_alias: bool = True,
    exclude_unset: bool = True,
) -> Self

BaseFirestoreModel | None

Parent document instance for subcollection models.

set | None

An explicit set of field names to include in the update. When provided, only those fields are written — all other exclusion rules still apply within the included set.

bool

default:"True"

Omit fields whose value is None.

bool

default:"True"

Use field aliases as Firestore field names.

bool

default:"True"

Omit fields that were not explicitly set on the instance.

Returns the same instance after the update.

user = await User.get("uid_abc123")
user.email = "[email protected]"
await user.update(include={"email"})

`delete`

Delete the document from Firestore. The instance must have a non-None id — raises ValueError if id is not set.

async def delete(self, cascade: bool = False) -> None

bool

default:"False"

When True, all subcollection documents under this document are recursively deleted before the document itself is removed. Requires subcollection model classes to be registered via init_firestore_odm.

Cascade deletion can be expensive for documents with large subcollections. It fetches every child document individually. Consider using Firestore’s built-in recursive delete for large datasets.

user = await User.get("uid_abc123")

# Delete only the user document
await user.delete()

# Delete the user and all their posts, comments, etc.
await user.delete(cascade=True)

`subcollection`

Return a SubCollectionAccessor bound to this instance and the given child model class, enabling Beanie-style subcollection query syntax.

def subcollection(self, child_cls: Type[BaseFirestoreModel]) -> SubCollectionAccessor

Type[BaseFirestoreModel]

required

The subcollection model class. Must declare Settings.parent = type(self) or a ValueError is raised.

user = await User.get("uid_abc123")
accessor = user.subcollection(Post)

async for post in accessor.find([Post.published == True]):
    print(post.title)

See the SubCollectionAccessor reference for the full accessor API.

Properties

`collection_name`

Return the Firestore collection name for the instance’s class, as declared in Settings.name. Falls back to the class name if Settings.name is not defined.

@property
def collection_name(self) -> str

user = User(username="alice", email="[email protected]")
print(user.collection_name)  # "users"

`get_collection_name`

Class-level counterpart to the collection_name property. Returns the same value but can be called on the class itself without an instance — useful inside other classmethods and internal path resolution helpers.

@classmethod
def get_collection_name(cls) -> str

print(User.get_collection_name())   # "users"
print(Post.get_collection_name())   # "posts"

Classes

Enums & Fields

BaseFirestoreModel — Async Firestore ODM Base Class

Class overview

The `Settings` inner class

The `id` field

Class methods

`initialize_db`

`initialize_fields`

`get`

`exists`

`find`

`find_one`

`count`

`batch_write`

`collection_group_find`

Instance methods

`save`

`update`

`delete`

`subcollection`

Properties

`collection_name`

`get_collection_name`

​Class overview

​The Settings inner class

​The id field

​Class methods

​initialize_db

​initialize_fields

​get

​exists

​find

​find_one

​count

​batch_write

​collection_group_find

​Instance methods

​save

​update

​delete

​subcollection

​Properties

​collection_name

​get_collection_name

Class overview

The `Settings` inner class

The `id` field

Class methods

`initialize_db`

`initialize_fields`

`get`

`exists`

`find`

`find_one`

`count`

`batch_write`

`collection_group_find`

Instance methods

`save`

`update`

`delete`

`subcollection`

Properties

`collection_name`

`get_collection_name`