FastAPI + dependency injection

Descripción

FastAPI tiene un sistema de dependency injection elegante y poderoso. La idea: en lugar de crear sessions de DB manualmente en cada endpoint, declaras db: AsyncSession = Depends(get_db) y FastAPI gestiona el ciclo de vida automáticamente.

Esta cápsula te enseña a configurar get_db() correctamente, integrarla con tu app FastAPI, y entender el patrón completo: una session por request, transacción automática, cleanup garantizado. También verás cómo extender el patrón para otras dependencies (auth, request context, etc.).

El problema sin dependency injection

Sin Depends, tendrías algo así:

# ❌ Patrón mal: session manual en cada endpoint
@app.get("/users/{username}")
async def get_user(username: str):
    async with AsyncSessionLocal() as session:
        try:
            user = (await session.execute(
                select(User).where(User.username == username)
            )).scalar_one_or_none()
            
            if not user:
                raise HTTPException(404)
            
            return user
        finally:
            await session.close()

Problemas:

• Boilerplate en cada endpoint
• Olvidar close() filtrando conexiones
• Difícil de testear (mockear la session es complicado)

Con Depends:

# ✅ Patrón correcto
@app.get("/users/{username}")
async def get_user(username: str, db: AsyncSession = Depends(get_db)):
    user = (await db.execute(select(User).where(User.username == username))).scalar_one_or_none()
    if not user:
        raise HTTPException(404)
    return user

FastAPI gestiona la session automáticamente.

Crear app/dependencies.py

# app/dependencies.py
"""Dependencies inyectables de FastAPI."""

from collections.abc import AsyncGenerator

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import AsyncSessionLocal


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """
    Provee una AsyncSession por request.
    
    - Se crea al inicio del request
    - Se cierra al final (success o error)
    - NO hace commit automático: ese es responsabilidad del endpoint
    """
    async with AsyncSessionLocal() as session:
        yield session
        # session.close() automático al salir del with

Importante: get_db() no hace commit. El endpoint decide cuándo commitear (típicamente al final si todo fue bien, vía await db.commit()).

Variante con commit/rollback automático

Algunos prefieren que get_db() maneje las transacciones:

async def get_db_transactional() -> AsyncGenerator[AsyncSession, None]:
    """Variante con auto-commit/rollback."""
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

Ventaja: nunca olvidas commit. Desventaja: pierdes control fino (a veces quieres commit a mitad del request).

Recomendación: usa la versión simple (get_db) y commit explícito en cada endpoint. Más predecible.

App entry point: app/main.py

# app/main.py
"""FastAPI app principal."""

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.routers import users, posts, comments


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Startup y shutdown hooks."""
    # Startup
    print("[startup] App iniciando...")
    yield
    # Shutdown
    print("[shutdown] App cerrando...")
    # Aquí cerrarías recursos: redis, kafka producers, etc.


app = FastAPI(
    title="Blog API",
    description="API del blog del path Backend Python Developer",
    version="1.0.0",
    lifespan=lifespan,
)


# Routers
app.include_router(users.router, prefix="/users", tags=["users"])
app.include_router(posts.router, prefix="/posts", tags=["posts"])
app.include_router(comments.router, prefix="/comments", tags=["comments"])


@app.get("/health")
async def healthcheck() -> dict[str, str]:
    """Endpoint de health check para load balancers."""
    return {"status": "ok"}

Estructura de routers

Crea app/routers/__init__.py (vacío) y app/routers/users.py:

# app/routers/users.py
"""Endpoints de users."""

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import get_db
from app.repositories import users as users_repo

router = APIRouter()


@router.get("/{username}")
async def get_user(
    username: str,
    db: AsyncSession = Depends(get_db),
):
    """Perfil de usuario por username."""
    user = await users_repo.get_by_username(db, username)
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"User '{username}' not found",
        )
    return user

Levantar el server

# Desarrollo: con auto-reload
uvicorn app.main:app --reload --port 8000

# Producción: con workers (cada worker = un proceso)
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4

Output:

INFO:     Uvicorn running on http://127.0.0.1:8000
INFO:     [startup] App iniciando...
INFO:     Application startup complete.

Visita:

• http://localhost:8000/health — endpoint de health check
• http://localhost:8000/docs — Swagger UI automático
• http://localhost:8000/redoc — ReDoc automático
• http://localhost:8000/openapi.json — OpenAPI schema

Probar el primer endpoint

Con httpie o curl:

curl http://localhost:8000/users/maria

Output (Maria del seed del módulo 2):

{
  "id": "a3e2f1d4-1b2c-4d5e-...",
  "email": "maria@blog.local",
  "username": "maria",
  "full_name": "María García",
  "bio": "Escritora de tecnología...",
  ...
}

Pero ⚠️ esto retorna toda la tabla (incluyendo password_hash). En la cápsula 06 corregimos eso con Pydantic schemas que controlan qué se serializa.

User no existente:

curl -i http://localhost:8000/users/no_existe
# HTTP/1.1 404 Not Found
# {"detail":"User 'no_existe' not found"}

El ciclo de vida de una request con Depends

1. Cliente HTTP envía: GET /users/maria
2. FastAPI matchea la ruta y descubre dependencies
3. FastAPI llama get_db():
   - Entra al "async with" de AsyncSessionLocal()
   - Hace yield session → la pasa al endpoint
4. El endpoint corre: get_user(username="maria", db=session)
5. El endpoint retorna o lanza
6. FastAPI vuelve al get_db():
   - El "async with" se cierra: session.close()
   - Si hubo excepción, propaga
7. Si no hubo error, FastAPI serializa la respuesta

Lo importante: una session por request. Cada request HTTP independiente, con su propia transacción.

Dependencies anidadas

Una dependency puede depender de otras. Ejemplo: validar que el user logueado existe.

# app/dependencies.py (extendido — esto es lookahead a guía #9)

from fastapi import Header

async def get_current_user_optional(
    authorization: str | None = Header(None),
    db: AsyncSession = Depends(get_db),
) -> User | None:
    """Lee el header Authorization y retorna el user o None."""
    if not authorization or not authorization.startswith("Bearer "):
        return None
    token = authorization[7:]
    # ... decodificar JWT y buscar user ...
    return user


async def get_current_user(
    user: User | None = Depends(get_current_user_optional),
) -> User:
    """Versión que requiere user logueado."""
    if user is None:
        raise HTTPException(status_code=401, detail="Not authenticated")
    return user

Auth completo es la guía #9 del path. Aquí solo demostramos que dependencies son composables.

Uso:

@router.post("/posts/")
async def create_post(
    payload: PostCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    # current_user ya está autenticado
    return await posts_repo.create(db, author_id=current_user.id, **payload.dict())

FastAPI resuelve las dependencies automáticamente, en orden, sin que tengas que pensar.

Dependencies como factory: Annotated

Python 3.12 hace muy elegante el patrón. En lugar de:

async def endpoint(db: AsyncSession = Depends(get_db)):
    ...

Puedes definir el alias:

# app/dependencies.py
from typing import Annotated
from fastapi import Depends

DbDep = Annotated[AsyncSession, Depends(get_db)]
CurrentUserDep = Annotated[User, Depends(get_current_user)]

Y usar:

async def endpoint(db: DbDep, user: CurrentUserDep):
    ...

Más limpio. Funcionalmente idéntico.

Healthcheck con DB

Mejorar el healthcheck para verificar que la DB responde:

# app/main.py
from sqlalchemy import text

@app.get("/health")
async def healthcheck(db: AsyncSession = Depends(get_db)) -> dict[str, str]:
    try:
        await db.execute(text("SELECT 1"))
        return {"status": "ok", "db": "ok"}
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"DB unavailable: {e}")

Útil para load balancers que verifican antes de rutear tráfico.

Errores comunes con dependency injection

Olvidar Depends(get_db)

# ❌ FALLA
async def endpoint(db: AsyncSession):
    pass
# Pydantic intenta validar AsyncSession como modelo

Solución: siempre Depends(...) con anotaciones para inyección.

Cerrar la session manualmente

async def endpoint(db: AsyncSession = Depends(get_db)):
    # ...
    await db.close()  # ❌ NO — get_db() se encarga
    return ...

get_db() cierra automáticamente en el yield. Cerrar manualmente puede causar errores raros.

Usar Depends en __init__ o middleware

Depends solo funciona en endpoints (path operations). Para middleware o startup, accede directo a AsyncSessionLocal().

Ejercicios

Ejercicio 1. Crea app/main.py con un endpoint /health que verifique la DB. Levanta uvicorn y haz curl.

# app/main.py
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import get_db

app = FastAPI(title="Blog API")


@app.get("/health")
async def healthcheck(db: AsyncSession = Depends(get_db)) -> dict[str, str]:
    try:
        await db.execute(text("SELECT 1"))
        return {"status": "ok", "db": "ok"}
    except Exception as e:
        raise HTTPException(503, f"DB unavailable: {e}")

uvicorn app.main:app --reload
# Otra terminal:
curl http://localhost:8000/health
# {"status":"ok","db":"ok"}

Ejercicio 2. Crea el router app/routers/users.py con un endpoint GET /users/{username} que use get_db. Inclúyelo en main.py.

curl http://localhost:8000/users/maria
# {"id": "...", "username": "maria", ...}

curl -i http://localhost:8000/users/no_existe
# HTTP/1.1 404 Not Found

Ejercicio 3. Convierte el repository users_repo.get_by_username del módulo 4 a async. Verifica que el endpoint sigue funcionando.

# app/repositories/users.py
from sqlalchemy.ext.asyncio import AsyncSession

async def get_by_username(session: AsyncSession, username: str) -> User | None:
    return (await session.execute(
        select(User).where(
            User.username == username,
            User.deleted_at.is_(None),
        )
    )).scalar_one_or_none()

Ejercicio 4. Crea un Annotated shortcut para db: AsyncSession = Depends(get_db) y usalo en el endpoint.

# app/dependencies.py
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

DbDep = Annotated[AsyncSession, Depends(get_db)]

# app/routers/users.py
from app.dependencies import DbDep

@router.get("/{username}")
async def get_user(username: str, db: DbDep):
    user = await users_repo.get_by_username(db, username)
    if user is None:
        raise HTTPException(404, f"User '{username}' not found")
    return user

Ejercicio 5. Levanta el server y abre /docs en el navegador. Explora la documentación interactiva. ¿Qué endpoints aparecen?

uvicorn app.main:app --reload

Resumen

• Depends(get_db) inyecta una AsyncSession por request
• get_db() con yield y async with maneja el ciclo de vida
• Una session por request — el patrón canónico
• Commit explícito en endpoints (no en get_db)
• include_router organiza endpoints en módulos
• Lifespan (startup/shutdown) para recursos compartidos
• /docs y /redoc se generan automáticamente
• Annotated shortcut para dependencies repetidas
• Healthcheck que verifica DB es buena práctica

En la siguiente cápsula construimos los endpoints CRUD completos.

Recursos Adicionales

1. FastAPI Dependencies — Documentación oficial
2. FastAPI Bigger Applications — Estructura con routers
3. FastAPI Lifespan Events — Startup/shutdown
4. SQLAlchemy + FastAPI Tutorial — Tutorial oficial

Siguiente: Cápsula 04 — Endpoints CRUD con repositorios async.