Skip to content

Common Patterns

Reusable code shapes built on BaseRepository and friends.

Pattern 1: Service Layer with Repository

Keep orchestration out of routes. A service composes one or more repositories and owns the domain rules.

Python
from mongo_ops import BaseRepository, BaseDocument


class User(BaseDocument):
    username: str = ""
    email: str = ""


class UserRepository(BaseRepository[User]):
    def __init__(self):
        super().__init__("users", User)

    async def find_by_email(self, email: str) -> User | None:
        doc = await self.collection.find_one({"email": email})
        return self.model(**doc) if doc else None


class UserService:
    def __init__(self, user_repo: UserRepository):
        self.user_repo = user_repo

    async def register_user(self, username: str, email: str) -> User:
        if await self.user_repo.find_by_email(email):
            raise ValueError("Email already exists")
        return await self.user_repo.create(User(username=username, email=email))

Register a unique index on email via ModelRegistry so the race is also caught by DuplicateKeyError — see use case 13.

Pattern 2: Aggregation Pipeline

Aggregations hit the raw Motor collection — wrap them in a repository method so callers stay at the domain level.

Python
from mongo_ops import BaseRepository, BaseDocument


class Post(BaseDocument):
    author_id: str = ""
    title: str = ""
    likes: int = 0


class PostRepository(BaseRepository[Post]):
    def __init__(self):
        super().__init__("posts", Post)

    async def get_user_stats(self, user_id: str) -> dict:
        pipeline = [
            {"$match": {"author_id": user_id}},
            {"$group": {"_id": None, "total_posts": {"$sum": 1}, "total_likes": {"$sum": "$likes"}}},
        ]
        result = await self.collection.aggregate(pipeline).to_list(1)
        return result[0] if result else {"total_posts": 0, "total_likes": 0}

Pattern 3: Bulk Operations

Multi-document mutation in one call — again wrapped at the repository boundary.

Python
from datetime import datetime

from bson import ObjectId
from mongo_ops import BaseRepository, BaseDocument


class Task(BaseDocument):
    status: str = "pending"


class TaskRepository(BaseRepository[Task]):
    def __init__(self):
        super().__init__("tasks", Task)

    async def bulk_update_status(self, ids: list[str], status: str) -> int:
        object_ids = [ObjectId(i) for i in ids]
        result = await self.collection.update_many(
            {"_id": {"$in": object_ids}},
            {"$set": {"status": status, "updated_at": datetime.utcnow()}},
        )
        return result.modified_count

Bulk/aggregation methods bypass the caching and population layers — keep them deliberate and documented.

Pattern 4: Transactional Multi-Step Write

Cross-collection atomicity with execute_transaction — see use case 12 for the full code.

Python
results = await TransactionManager.execute_transaction([op1, op2, op3])
# op1/op2/op3 each receive the shared session; one failure rolls back all.