Transactions

Use Case 3: Transaction Support for Multi-Document Operations

Scenario: Order processing system that updates inventory and creates order atomically.

Python

from mongo_ops import BaseDocument, BaseRepository, TransactionManager
from datetime import datetime
from typing import List
from pydantic import Field

class Order(BaseDocument):
    user_id: str
    items: List[dict]  # [{"product_id": "...", "quantity": 2}]
    total_amount: float
    status: str = "pending"

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

class OrderService:
    def __init__(self, order_repo: OrderRepository, product_repo: ProductRepository):
        self.order_repo = order_repo
        self.product_repo = product_repo

    async def create_order_with_inventory_update(self, order: Order) -> Order:
        """Create order and update inventory atomically"""

        async def create_order_transaction(session):
            # Insert order
            order_doc = order.model_dump(exclude={"id"}, exclude_none=True)
            order_doc["created_at"] = datetime.utcnow()
            order_doc["updated_at"] = datetime.utcnow()
            result = await self.order_repo.collection.insert_one(order_doc, session=session)

            # Update inventory for each item
            for item in order.items:
                await self.product_repo.collection.update_one(
                    {"_id": ObjectId(item["product_id"])},
                    {
                        "$inc": {"quantity": -item["quantity"]},
                        "$set": {"updated_at": datetime.utcnow()}
                    },
                    session=session
                )

            # Fetch created order
            created = await self.order_repo.collection.find_one(
                {"_id": result.inserted_id},
                session=session
            )
            return Order(**created)

        # Execute in transaction
        async with TransactionManager.start_session() as session:
            return await create_order_transaction(session)

# Usage in FastAPI
@app.post("/orders/", response_model=Order)
async def create_order(order: Order):
    order_service = OrderService(order_repo, product_repo)
    try:
        return await order_service.create_order_with_inventory_update(order)
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))