Beanie ODM for MongoDB

This skill provides patterns for MongoDB integration using Beanie ODM with async Motor driver, optimized for FastAPI applications.

Database Initialization

Connection Setup

from beanie import init_beanie from motor.motor_asyncio import AsyncIOMotorClient from app.domains.users.models import User from app.domains.products.models import Product

async def init_database(settings: Settings): client = AsyncIOMotorClient(settings.mongodb_url)

await init_beanie(
    database=client[settings.database_name],
    document_models=[
        User,
        Product,
        # Add all document models
    ]
)

Settings Configuration

from pydantic_settings import BaseSettings

class Settings(BaseSettings): mongodb_url: str = "mongodb://localhost:27017" database_name: str = "app_db"

class Config:
    env_file = ".env"

Document Models

Basic Document

from beanie import Document, Indexed from pydantic import Field, EmailStr from datetime import datetime from typing import Optional

class User(Document): email: Indexed(EmailStr, unique=True) name: str hashed_password: str is_active: bool = True created_at: datetime = Field(default_factory=datetime.utcnow) updated_at: datetime = Field(default_factory=datetime.utcnow)

class Settings:
    name = "users"  # Collection name
    use_state_management = True

class Config:
    json_schema_extra = {
        "example": {
            "email": "user@example.com",
            "name": "John Doe"
        }
    }

Document with Relationships

from beanie import Document, Link, BackLink from typing import List, Optional

class Author(Document): name: str books: List[BackLink["Book"]] = Field(original_field="author")

class Settings:
    name = "authors"

class Book(Document): title: str author: Link[Author] categories: List[Link["Category"]] = []

class Settings:
    name = "books"

class Category(Document): name: str books: List[BackLink[Book]] = Field(original_field="categories")

class Settings:
    name = "categories"

Embedded Documents

from beanie import Document from pydantic import BaseModel from typing import List

class Address(BaseModel): street: str city: str country: str postal_code: str

class Contact(BaseModel): type: str # "email", "phone" value: str is_primary: bool = False

class Customer(Document): name: str addresses: List[Address] = [] contacts: List[Contact] = []

class Settings:
    name = "customers"

Query Patterns

Basic CRUD Operations

Create

user = User(email="user@example.com", name="John") await user.insert()

Create with validation

user = await User.insert_one( User(email="user@example.com", name="John") )

Read by ID

user = await User.get(user_id)

Read with filter

users = await User.find(User.is_active == True).to_list()

Update

user.name = "Jane" await user.save()

Partial update

await user.set({User.name: "Jane", User.updated_at: datetime.utcnow()})

Delete

await user.delete()

Advanced Queries

from beanie.operators import In, RegEx, And, Or

Find with operators

active_users = await User.find( And( User.is_active == True, User.created_at >= start_date ) ).to_list()

Regex search

users = await User.find( RegEx(User.name, "^John", options="i") ).to_list()

In operator

users = await User.find( In(User.email, ["a@test.com", "b@test.com"]) ).to_list()

Pagination

users = await User.find_all().skip(20).limit(10).to_list()

Sorting

users = await User.find_all().sort(-User.created_at).to_list()

Projection (select specific fields)

users = await User.find_all().project(UserSummary).to_list()

Aggregation Pipelines

from beanie import PydanticObjectId

class UserStats(BaseModel): total_users: int active_users: int avg_age: float

Aggregation pipeline

pipeline = [ {"$match": {"is_active": True}}, {"$group": { "_id": None, "total": {"$sum": 1}, "avg_age": {"$avg": "$age"} }} ]

result = await User.aggregate(pipeline).to_list()

Using Beanie aggregation

from beanie.odm.queries.aggregation import AggregationQuery

stats = await User.find(User.is_active == True).aggregate([ {"$group": { "_id": "$department", "count": {"$sum": 1} }} ]).to_list()

Indexes

from beanie import Document, Indexed from pymongo import IndexModel, ASCENDING, DESCENDING, TEXT

class Product(Document): # Single field index sku: Indexed(str, unique=True)

# Compound index defined in Settings
name: str
category: str
price: float
description: str

class Settings:
    name = "products"
    indexes = [
        # Compound index
        IndexModel(
            [("category", ASCENDING), ("price", DESCENDING)],
            name="category_price_idx"
        ),
        # Text index
        IndexModel(
            [("name", TEXT), ("description", TEXT)],
            name="search_idx"
        ),
        # TTL index
        IndexModel(
            [("expires_at", ASCENDING)],
            expireAfterSeconds=0,
            name="ttl_idx"
        )
    ]

Transactions

from beanie import Document from motor.motor_asyncio import AsyncIOMotorClientSession

async def transfer_funds( from_account_id: str, to_account_id: str, amount: float, session: AsyncIOMotorClientSession ): async with await session.start_transaction(): from_account = await Account.get(from_account_id, session=session) to_account = await Account.get(to_account_id, session=session)

    if from_account.balance &#x3C; amount:
        raise ValueError("Insufficient funds")

    await from_account.set(
        {Account.balance: from_account.balance - amount},
        session=session
    )
    await to_account.set(
        {Account.balance: to_account.balance + amount},
        session=session
    )

Service Layer Pattern

from typing import List, Optional from beanie import PydanticObjectId

class UserService: async def get_by_id(self, user_id: str) -> Optional[User]: return await User.get(PydanticObjectId(user_id))

async def get_by_email(self, email: str) -> Optional[User]:
    return await User.find_one(User.email == email)

async def get_all(
    self,
    skip: int = 0,
    limit: int = 100,
    is_active: Optional[bool] = None
) -> List[User]:
    query = User.find_all()
    if is_active is not None:
        query = User.find(User.is_active == is_active)
    return await query.skip(skip).limit(limit).to_list()

async def create(self, data: UserCreate) -> User:
    user = User(**data.model_dump())
    await user.insert()
    return user

async def update(self, user_id: str, data: UserUpdate) -> Optional[User]:
    user = await self.get_by_id(user_id)
    if not user:
        return None
    update_data = data.model_dump(exclude_unset=True)
    update_data["updated_at"] = datetime.utcnow()
    await user.set(update_data)
    return user

Additional Resources

Reference Files

For detailed patterns and migration guides:

references/migrations.md
Database migration strategies
references/performance.md
Query optimization tips
references/relationships.md
Link and BackLink patterns

Example Files

Working examples in examples/ :

examples/document_models.py
Complete document definitions
examples/aggregations.py
Aggregation pipeline examples
examples/service.py
Service layer implementation

beanie odm

Safety Notice

Copy this and send it to your AI assistant to learn