Python FastAPI dependency injection for database transactions

from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import async_session_factory

async def get_db():
    """Provides a session with auto-commit on success, rollback on error."""
    async with async_session_factory() as session:
        async with session.begin():
            yield session
        # Commits on success, rolls back on exception

DbSession = Annotated[AsyncSession, Depends(get_db)]

# For manual transaction control:
async def get_db_manual():
    """Provides a session without auto-commit -- caller controls transaction."""
    async with async_session_factory() as session:
        yield session

# Route that uses auto-commit:
@router.post('/votes')
async def cast_vote(vote: VoteCreate, db: DbSession):
    new_vote = Vote(**vote.model_dump())
    db.add(new_vote)
    # No commit needed -- session.begin() handles it
    return new_vote

# Route with multiple dependent operations:
@router.post('/traces/{trace_id}/validate')
async def validate_trace(trace_id: str, db: DbSession):
    trace = await db.get(Trace, trace_id)
    if not trace:
        raise HTTPException(404)

    trace.status = 'validated'
    trace.trust_score = 1.0

    # Update contributor stats in same transaction:
    await update_contributor_stats(db, trace.contributor_id)

    # Both updates commit together or both roll back