Skip to content

Use Case 18: Aggregation Pipelines

Scenario: A reporting dashboard needs grouped statistics, computed fields, and joined data — operations that don't map to single-document CRUD.

📦 What's New?

Component Description
collection.aggregate() Raw Motor aggregation — no base-repo helper exists for pipelines.
Repository wrapper Keeps pipeline logic behind a domain method so callers never touch the collection directly.
$group, $lookup, $project Common pipeline stages demonstrated in a single repository.

🚀 Example

Python
from fastapi import FastAPI
from mongo_ops import BaseDocument, BaseRepository


# ---------------------------
# Models
# ---------------------------
class Order(BaseDocument):
    user_id: str
    status: str = "pending"
    total: float = 0.0


class OrderRepository(BaseRepository[Order]):
    def __init__(self):
        super().__init__("orders", Order)

    async def revenue_by_status(self) -> list[dict]:
        """Group orders by status, sum totals."""
        pipeline = [
            {"$group": {"_id": "$status", "revenue": {"$sum": "$total"}, "count": {"$sum": 1}}},
            {"$sort": {"revenue": -1}},
        ]
        return await self.collection.aggregate(pipeline).to_list(length=100)

    async def top_spenders(self, limit: int = 10) -> list[dict]:
        """Top N users by total spend, with a lookup to resolve user names."""
        pipeline = [
            {"$group": {"_id": "$user_id", "total_spent": {"$sum": "$total"}, "order_count": {"$sum": 1}}},
            {"$sort": {"total_spent": -1}},
            {"$limit": limit},
            {"$lookup": {
                "from": "users",
                "localField": "_id",
                "foreignField": "_id",
                "as": "user",
            }},
            {"$unwind": {"path": "$user", "preserveNullAndEmptyArrays": True}},
            {"$project": {
                "total_spent": 1,
                "order_count": 1,
                "user_name": {"$ifNull": ["$user.username", "unknown"]},
            }},
        ]
        return await self.collection.aggregate(pipeline).to_list(length=limit)

    async def daily_summary(self, days: int = 30) -> list[dict]:
        """Order counts grouped by day for the last N days."""
        from datetime import datetime, timedelta

        cutoff = datetime.utcnow() - timedelta(days=days)
        pipeline = [
            {"$match": {"created_at": {"$gte": cutoff}}},
            {"$group": {
                "_id": {"$dateToString": {"format": "%Y-%m-%d", "date": "$created_at"}},
                "orders": {"$sum": 1},
                "revenue": {"$sum": "$total"},
            }},
            {"$sort": {"_id": 1}},
        ]
        return await self.collection.aggregate(pipeline).to_list(length=days)


app = FastAPI()
order_repo = OrderRepository()


@app.get("/reports/revenue-by-status")
async def revenue_by_status():
    return await order_repo.revenue_by_status()


@app.get("/reports/top-spenders")
async def top_spenders(limit: int = 10):
    return await order_repo.top_spenders(limit)


@app.get("/reports/daily-summary")
async def daily_summary(days: int = 30):
    return await order_repo.daily_summary(days)

Note: lifespan wiring omitted — copy from use case 01.

💡 Tips

  • Aggregation methods bypass the caching and population layers — they read raw documents. Keep them deliberate and document the bypass in the method docstring.
  • Use $lookup sparingly; for read-heavy workloads, prefer application-level population via use case 08 which benefits from the cache.
  • Pipeline results are plain dicts, not model instances. Wrap them in a Pydantic model if you need validation or serialization in responses.
  • Create indexes on fields used in $match and $sort stages — otherwise the aggregation runs a collection scan (see use case 13).