🇺🇸English

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 < 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

Weekly Installs

–

Repository

lobbi-docs/claude

GitHub Stars

9

First Seen

–

Security Audits

Gen Agent Trust HubPassSocketPassSnykPass