Skip to content

Use Case 12: Using TransactionManager.execute_transaction

Scenario: Perform several writes across different collections atomically — e.g., create an Order and decrement Inventory. The low-level start_session context works, but execute_transaction collects results from a list of async operations.

📦 What's New?

Component Description
TransactionManager.execute_transaction Runs a list of async callables (each receives a session) inside one transaction; returns the result of each callable, in order.
Automatic rollback Any raised exception aborts the transaction and propagates to the caller.
start_session The underlying async context manager async with TransactionManager.start_session() as session:.

🚀 Example

Python
from contextlib import asynccontextmanager
from datetime import datetime

from bson import ObjectId
from fastapi import FastAPI, HTTPException
from mongo_ops import BaseDocument, ModelRegistry, MongoConnectionManager
from mongo_ops.repository import BaseRepository
from mongo_ops.transactions import TransactionManager


# 1. Models
class Order(BaseDocument):
    user_id: str = ""
    items: list[dict] = []  # [{"product_id": ObjectId, "qty": int}]
    total: float = 0.0


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


# 2. Repositories (constructed inside the lifespan — after connect()).
class OrderRepo(BaseRepository[Order]):
    def __init__(self):
        super().__init__("orders", Order)


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


# 3. The atomic operation.
async def create_order_with_inventory(order: Order, order_repo: OrderRepo, inv_repo: InventoryRepo):
    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["qty"]}},
                session=session,
            )
        return "inventory-updated"

    results = await TransactionManager.execute_transaction(
        [insert_order, update_inventory]
    )
    return results[0]  # created order document; results[1] == "inventory-updated"


# 4. FastAPI wiring.
app = FastAPI()


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


app.lifespan = lifespan


@app.post("/orders/", response_model=Order)
async def create_order(order: Order):
    created = await create_order_with_inventory(order, OrderRepo(), InventoryRepo())
    if created is None:
        raise HTTPException(status_code=400, detail="Transaction failed")
    return created

💡 Tips

  • Return values: each callable can return whatever you need; results are collected in the same order.
  • Errors: raise inside any callable → the whole transaction aborts (session rolls back) and the exception propagates.
  • Reads inside a transaction: pass session=session to find_one/find too.
  • Testing (no Mongo): monkeypatch a fake client on mongo_ops.transactions.MongoConnectionManager.get_client and stub start_session — see tests/test_transactions.py.
  • Repo models still carry created_at/updated_at; for raw collection inserts inside the transaction you set them manually (as shown).