Pydantic Mongo

To perform complex queries that go beyond standard CRUD operations, you can implement custom methods within your repository class. Use self.get_collection() to access the underlying PyMongo collection object, allowing you to run raw aggregation pipelines. Since aggregation results often have a different shape than your primary models, you should define a separate Pydantic BaseModel to represent the aggregation output and map the raw dictionary results to these models manually.

from pydantic import BaseModel
from pydantic_mongo import AbstractRepository

# 1. Define a model for the specific aggregation output
class UserStats(BaseModel):
    role: str
    average_age: float
    total_count: int

class UserRepository(AbstractRepository[User]):
    class Meta:
        collection_name = 'users'

    def get_user_stats(self) -> list[UserStats]:
        """Get statistics about users grouped by role."""
        pipeline = [
            {"$group": {
                "_id": "$role",
                "average_age": {"$avg": "$age"},
                "total_count": {"$sum": 1}
            }},
            {"$project": {
                "role": "$_id",
                "average_age": 1,
                "total_count": 1,
                "_id": 0
            }}
        ]
        # 2. Access the PyMongo collection
        collection = self.get_collection()
        # 3. Execute the aggregation
        results = collection.aggregate(pipeline)
        # 4. Map results to the custom Pydantic model
        return [UserStats(**doc) for doc in results]

# Usage
repo = UserRepository(database)
stats = repo.get_user_stats()
for stat in stats:
    print(f"Role: {stat.role}, Avg Age: {stat.average_age}, Count: {stat.total_count}")

You can extend a repository by defining custom methods within your AbstractRepository subclass. This allows you to combine high-level Pydantic-validated operations with low-level PyMongo operations.

Key tools for extending functionality:

• get_collection(): Returns the underlying PyMongo Collection object, enabling any raw MongoDB operation (like update_many, aggregate, or find_one with specific projections).
• to_model(data): Converts raw PyMongo dictionary output into your defined Pydantic model, automatically handling the mapping of MongoDB's _id to your model's id field.
• to_model_custom(model_class, data): Converts raw PyMongo output into a different Pydantic model than the one the repository is primarily typed for. This is useful for projections or specialized views.
• find_by(...): Use this within custom methods to perform standard queries while maintaining the repository's return type.

class UserRepository(AbstractRepository[User]):
    class Meta:
        collection_name = 'users'

    def find_users_by_domain(self, domain: str) -> Iterable[User]:
        """Custom method using standard find_by."""
        return self.find_by({"email": {"$regex": f"@{domain}$"}})

    def bulk_update_status(self, status: str, user_ids: list):
        """Custom method using raw PyMongo via get_collection()."""
        collection = self.get_collection()
        collection.update_many(
            {"_id": {"$in": user_ids}},
            {"$set": {"status": status}}
        )

    def get_user_by_raw_query(self, user_id: str) -> Optional[User]:
        """Custom method fetching raw data and converting to the primary model."""
        collection = self.get_collection()
        raw_data = collection.find_one({"_id": ObjectId(user_id)})
        return self.to_model(raw_data) if raw_data else None

    def get_custom_user_view(self, user_id: str) -> Optional[UserProfile]:
        """Custom method fetching raw data and converting to a different model (projection)."""
        collection = self.get_collection()
        raw_data = collection.find_one(
            {"_id": ObjectId(user_id)},
            projection={"name": 1, "age": 1, "_id": 0}
        )
        return self.to_model_custom(UserProfile, raw_data) if raw_data else None

To use pydantic-mongo, you must define your data models using Pydantic's BaseModel and create a repository class by inheriting from AbstractRepository (for synchronous use) or AsyncAbstractRepository (for asynchronous use).

Use PydanticObjectId for the id field in your models; it is an alias to Annotated[ObjectId, ObjectIdAnnotation] that ensures compatibility with MongoDB's ObjectId.

In your repository class, define a Meta inner class with a collection_name to specify which MongoDB collection the repository manages.

from bson import ObjectId
from pydantic import BaseModel
from pydantic_mongo import AbstractRepository, PydanticObjectId
from pymongo import MongoClient
from typing import Optional, List

# 1. Define your models
class Foo(BaseModel):
   count: int
   size: float = None

class Spam(BaseModel):
   # PydanticObjectId handles MongoDB ObjectId conversion
   id: Optional[PydanticObjectId] = None
   foo: Foo
   bars: List[Bar]

# 2. Create a repository
class SpamRepository(AbstractRepository[Spam]):
   class Meta:
      collection_name = 'spams'

# 3. Connect to database and initialize
client = MongoClient("mongodb://localhost:27017")
database = client["example"]
repo = SpamRepository(database)

To perform asynchronous MongoDB operations, use AsyncMongoClient from pymongo and inherit from AsyncAbstractRepository when defining your repository. Your repository class must define a Meta inner class with the collection_name attribute.

When using the async repository, all database interaction methods (such as .save() and .find_one_by_id()) must be awaited.

from pymongo import AsyncMongoClient
from pydantic import BaseModel
from pydantic_mongo import AsyncAbstractRepository

class User(BaseModel):
    id: str
    name: str
    email: str

class UserRepository(AsyncAbstractRepository[User]):
    class Meta:
        collection_name = 'users'

# Initialize database connection
client = AsyncMongoClient('mongodb://localhost:27017')
database = client["mydb"]

# Create repository instance
user_repo = UserRepository(database)

# Example usage
async def create_user():
    user = User(name='John Doe', email='john@example.com')
    await user_repo.save(user)

async def find_user(user_id: str):
    user = await user_repo.find_one_by_id(user_id)
    return user