Skip to content

Use Case 2: Custom Repository with Business Logic

Scenario: An e-commerce product catalog needs search, filtering, and stock updates without Mongo leaking into routes.

📦 What's New?

Component Description
Repository methods Encapsulate queries ($regex, filters, $inc) behind domain methods.
get_many Filtering + default pagination via the base repository.
Direct collection access For operations with no base-repo helper (regex search, atomic $inc).

🚀 Example

Python
from fastapi import FastAPI, HTTPException, Query
from mongo_ops import BaseDocument, BaseRepository

# ---------------------------
# Model
# ---------------------------
class Product(BaseDocument):
    name: str
    description: str = ""
    price: float = 0.0
    category: str = ""
    in_stock: bool = True
    quantity: int = 0
    tags: list[str] = []


# ---------------------------
# Repository
# ---------------------------
class ProductRepository(BaseRepository[Product]):
    def __init__(self):
        super().__init__("products", Product)

    async def search_by_name(self, query: str) -> list[Product]:
        """Case-insensitive name search."""
        docs = await self.collection.find(
            {"name": {"$regex": query, "$options": "i"}}
        ).to_list(length=100)
        return [self.model(**doc) for doc in docs]

    async def get_by_category(self, category: str, in_stock_only: bool = True) -> list[Product]:
        filter_query = {"category": category}
        if in_stock_only:
            filter_query["in_stock"] = True
        return await self.get_many(filter=filter_query)

    async def get_low_stock(self, threshold: int = 10) -> list[Product]:
        return await self.get_many(filter={"quantity": {"$lt": threshold}, "in_stock": True})

    async def update_stock(self, product_id: str, quantity_delta: int) -> Product | None:
        """Atomically increment/decrement stock."""
        from bson import ObjectId
        from datetime import datetime

        result = await self.collection.find_one_and_update(
            {"_id": ObjectId(product_id)},
            {"$inc": {"quantity": quantity_delta}, "$set": {"updated_at": datetime.utcnow()}},
            return_document=True,
        )
        return self.model(**result) if result else None


app = FastAPI()
product_repo = ProductRepository()


@app.get("/products/search", response_model=list[Product])
async def search_products(q: str = Query(..., min_length=1)):
    return await product_repo.search_by_name(q)


@app.get("/products/category/{category}", response_model=list[Product])
async def products_by_category(category: str, in_stock: bool = True):
    return await product_repo.get_by_category(category, in_stock)


@app.get("/products/low-stock", response_model=list[Product])
async def low_stock_products(threshold: int = 10):
    return await product_repo.get_low_stock(threshold)


@app.patch("/products/{product_id}/stock")
async def update_product_stock(product_id: str, quantity_delta: int):
    product = await product_repo.update_stock(product_id, quantity_delta)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return product

Note: this snippet omits the FastAPI lifespan connection wiring for brevity — copy it from use case 01 so ProductRepository() is created only after MongoConnectionManager.connect().

💡 Tips

  • Methods that hit self.collection directly (regex search, $inc) bypass the caching and population layers. If a feature composes them — extend CachedBaseRepository or PopulatingRepository instead and add the domain methods there.
  • Prefer get_many(filter=...) over raw find when you want pagination/sort defaults for free.
  • Reuse self.model(**doc) to convert raw dicts to model instances consistently.