Skip to main content
SubCollectionAccessor provides a concise, parent-scoped interface for querying and mutating subcollection documents. You obtain an instance by calling parent_instance.subcollection(ChildModel) rather than constructing it directly. Every operation the accessor exposes is equivalent to calling the corresponding class method on the child model with parent=parent_instance, but the Beanie-inspired fluent style makes subcollection code easier to read.

​
Obtaining an accessor

Call subcollection() on any hydrated BaseFirestoreModel instance and pass the child model class: 
user = await User.get("uid_abc123")
posts = user.subcollection(Post)

async for post in posts.find():
    print(post.title)
subcollection() immediately validates that Post.Settings.parent equals type(user) (i.e. User). A ValueError is raised if the child class does not declare the correct parent type — this catches misconfigured models at the earliest possible moment.
The child model class must declare Settings.parent = User (or whatever the parent type is). Passing a top-level collection model raises ValueError: Post does not declare Settings.parent = User.
class Post(BaseFirestoreModel):
    class Settings:
        name = "posts"
        parent = User       # required for SubCollectionAccessor to accept it

    title: str
    published: bool = False

​
Methods

​
add

Create a new document in the subcollection. Delegates to doc.save(parent=self._parent, **kwargs). 
async def add(self, doc: BaseFirestoreModel, **kwargs) -> BaseFirestoreModel
​
doc
BaseFirestoreModel
required
An unsaved model instance to persist. If doc.id is None, Firestore auto-generates a document ID.
​
**kwargs
Any
Additional keyword arguments forwarded to save(). Supported keys include exclude_none (default True), by_alias (default True), and exclude_unset (default True).
Returns the saved instance with id populated. 
post = Post(title="Hello World", published=True)
await user.subcollection(Post).add(post)
print(post.id)  # auto-generated

​
get

Retrieve a single subcollection document by its document ID. 
async def get(self, doc_id: str) -> Optional[BaseFirestoreModel]
​
doc_id
str
required
The Firestore document ID to look up within the subcollection.
Returns the hydrated child model instance, or None if no document with that ID exists in this parent’s subcollection. 
post = await user.subcollection(Post).get("pid_xyz")
if post:
    print(post.title)

​
find

Stream all documents in the subcollection that match the given filters. Returns an AsyncGenerator — iterate with async for. 
async def find(self, filters=None, **kwargs) -> AsyncGenerator
​
filters
List[Tuple[FieldType, FirestoreOperators, Any]] | None
Filter tuples to apply. Build them with FirestoreField expressions (Post.published == True) or raw (field, operator, value) tuples. Defaults to None (returns all documents).
​
**kwargs
Any
Additional keyword arguments forwarded to BaseFirestoreModel.find(). Supported keys include projection, order_by, limit, and offset.
from firestore_pydantic_odm import OrderByDirection

async for post in user.subcollection(Post).find(
    filters=[Post.published == True],
    order_by=(Post.created_at, OrderByDirection.DESCENDING),
    limit=10,
):
    print(post.title)

​
find_one

Return the first document in the subcollection that matches the given filters, or None if no match is found. 
async def find_one(self, filters=None, **kwargs) -> Optional[BaseFirestoreModel]
​
filters
List[Tuple[FieldType, FirestoreOperators, Any]] | None
Filter tuples. Passing None or [] returns the first document in the subcollection.
​
**kwargs
Any
Additional keyword arguments forwarded to BaseFirestoreModel.find_one(). Supported keys include projection and order_by.
latest_post = await user.subcollection(Post).find_one(
    filters=[Post.published == True],
    order_by=(Post.created_at, OrderByDirection.DESCENDING),
)

​
count

Return the number of documents in the subcollection that match the given filters. 
async def count(self, filters=None) -> int
​
filters
List[Tuple[FieldType, FirestoreOperators, Any]] | None
Filter tuples. Passing None counts all documents in the subcollection.
total = await user.subcollection(Post).count()
published = await user.subcollection(Post).count([Post.published == True])
print(f"{published}/{total} posts published")

​
exists

Return True if a document with the given ID exists in this parent’s subcollection. 
async def exists(self, doc_id: str) -> bool
​
doc_id
str
required
The Firestore document ID to check within the subcollection.
if await user.subcollection(Post).exists("pid_xyz"):
    print("Post exists")

​
delete

Delete a document from the subcollection. The document instance must have a non-None id. 
async def delete(self, doc: BaseFirestoreModel) -> None
​
doc
BaseFirestoreModel
required
A hydrated child model instance to delete. Delegates to doc.delete() — the cascade parameter from BaseFirestoreModel.delete is not exposed here; call doc.delete(cascade=True) directly if you need recursive deletion.
post = await user.subcollection(Post).get("pid_xyz")
if post:
    await user.subcollection(Post).delete(post)

​
Accessor vs. direct class methods

Both styles are fully equivalent. The accessor is a syntactic convenience — choose whichever reads more clearly in your codebase. 
user = await User.get("uid_abc123")
accessor = user.subcollection(Post)

# Create
new_post = Post(title="Using the accessor")
await accessor.add(new_post)

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

# Get by ID
post = await accessor.get("pid_xyz")

# Count
n = await accessor.count([Post.published == True])

# Delete
await accessor.delete(post)
The accessor style is especially readable when you make multiple calls against the same parent–child pair in one function, because you only reference the parent once when constructing the accessor.

​
Full subcollection example

from firestore_pydantic_odm import (
    BaseFirestoreModel,
    FirestoreDB,
    init_firestore_odm,
    OrderByDirection,
)

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

class Post(BaseFirestoreModel):
    class Settings:
        name = "posts"
        parent = User
    title: str
    published: bool = False

db = FirestoreDB(project_id="my-project")
init_firestore_odm(database=db, document_models=[User, Post])

async def demo():
    # Create a user
    user = User(username="alice")
    await user.save()

    # Add posts via the accessor
    posts_accessor = user.subcollection(Post)
    await posts_accessor.add(Post(title="Draft post"))
    await posts_accessor.add(Post(title="Published post", published=True))

    # Query only published posts
    async for post in posts_accessor.find(
        filters=[Post.published == True],
        order_by=(Post.title, OrderByDirection.ASCENDING),
    ):
        print(post.title)  # "Published post"

    # Count
    total = await posts_accessor.count()
    print(f"{total} total posts")  # 2

    # Clean up with cascade delete (removes user + all posts)
    await user.delete(cascade=True)