Pythonfrom mongo_ops import BaseDocument, BaseRepository
-from typing import Optional, List
-from pydantic import Field
-
-class Product(BaseDocument):
- name: str = Field(..., min_length=1)
- description: str
- price: float = Field(..., gt=0)
- category: str
- in_stock: bool = True
- quantity: int = Field(default=0, ge=0)
- tags: List[str] = []
-
-class ProductRepository(BaseRepository[Product]):
- def __init__(self):
- super().__init__("products", Product)
-
- async def search_by_name(self, query: str) -> List[Product]:
- """Search products by name (case-insensitive)"""
- 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]:
- """Get products by category"""
- 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]:
- """Get products with low stock"""
- return await self.get_many(
- filter={"quantity": {"$lt": threshold}, "in_stock": True}
- )
-
- async def update_stock(self, product_id: str, quantity_delta: int) -> Optional[Product]:
- """Update product stock (increment/decrement)"""
- 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
-
-# Usage in FastAPI
-from fastapi import FastAPI, Query
-
-app = FastAPI()
-product_repo = ProductRepository()
+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
+Pythonfrom 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.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 = FastAPI()
+product_repo = ProductRepository()
+
-@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/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/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
+
+@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.
+
+
+
+
diff --git a/mongo-ops/03_use_cases/03_transactions/index.html b/mongo-ops/03_use_cases/03_transactions/index.html
index e2e6a46..96da10e 100644
--- a/mongo-ops/03_use_cases/03_transactions/index.html
+++ b/mongo-ops/03_use_cases/03_transactions/index.html
@@ -457,7 +457,7 @@
- 02 components
+ Core Components
@@ -575,6 +575,8 @@
+
+