Skip to content

Use Case 3: Transaction Support for Multi-Document Operations

Scenario: Order processing must update inventory and create an order atomically. Any failure rolls both back.

📦 What's New?

Component Description
TransactionManager.start_session Async context manager yielding a session with an active transaction.
session= kwarg Pass to every insert_one / update_one / find_one inside the block.

🚀 Example

Python
from contextlib import asynccontextmanager
from datetime import datetime

from bson import ObjectId
from fastapi import FastAPI, HTTPException
from mongo_ops import BaseDocument, BaseRepository, MongoConnectionManager, TransactionManager


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


class Inventory(BaseDocument):
    product_id: ObjectId = None
    quantity: int = 0


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


class InventoryRepository(BaseRepository[Inventory]):
    def __init__(self):
        super().__init__("inventory", Inventory)


async def create_order_with_inventory_update(
    order: Order,
    order_repo: OrderRepository,
    inv_repo: InventoryRepository,
) -> Order:
    async def insert_order(session):
        doc = order.model_dump(exclude={"id"}, exclude_none=True)
        doc["created_at"] = doc["updated_at"] = datetime.utcnow()
        result = await order_repo.collection.insert_one(doc, session=session)
        return await order_repo.collection.find_one({"_id": result.inserted_id}, session=session)

    async def update_inventory(session):
        for item in order.items:
            await inv_repo.collection.update_one(
                {"product_id": ObjectId(item["product_id"])},
                {
                    "$inc": {"quantity": -item["quantity"]},
                    "$set": {"updated_at": datetime.utcnow()},
                },
                session=session,
            )
        return None

    async with TransactionManager.start_session() as session:
        created = await insert_order(session)
        await update_inventory(session)
        return Order(**created)


# ---------------------------
# FastAPI wiring
# ---------------------------
app = FastAPI()

order_repo = OrderRepository()
inv_repo = InventoryRepository()


@asynccontextmanager
async def lifespan(_app: FastAPI):
    async with MongoConnectionManager.lifespan(
        uri="mongodb://localhost:27017", db_name="shop"
    ):
        yield


app.lifespan = lifespan


@app.post("/orders/", response_model=Order)
async def create_order(order: Order):
    try:
        return await create_order_with_inventory_update(order, order_repo, inv_repo)
    except Exception as exc:
        raise HTTPException(status_code=400, detail=str(exc))

💡 Tips

  • Repositories are created inside the lifespan (or a dependency) so the database is connected — see the note in use case 01.
  • Every operation inside the transaction context must receive session=<session> — a missing session silently runs outside the transaction.
  • MongoDB transactions require a replica set (or the local standalone test server that emulates one).
  • For a list-of-operations style, see TransactionManager.execute_transaction in use case 12.