05 - Inyección de Dependencias en FastAPI

1. De NestJS a FastAPI: Cambio de Paradigma

En NestJS, la DI se basa en clases, decoradores y un contenedor IoC que instancia servicios automáticamente. En FastAPI, el sistema es funcional y explícito: usas funciones (o clases callable) que se resuelven en cada request.

// NestJS: Class-based DI con decoradores
@Injectable()
export class UsersService {
constructor(
  private readonly db: DatabaseService,
  private readonly cache: CacheService,
) {}

async findOne(id: string): Promise<User> {
  const cached = await this.cache.get(id);
  if (cached) return cached;
  return this.db.users.findUnique({ where: { id } });
}
}

@Controller('users')
export class UsersController {
constructor(private readonly usersService: UsersService) {}

@Get(':id')
findOne(@Param('id') id: string) {
  return this.usersService.findOne(id);
}
}

# FastAPI: Functional DI con Depends
from fastapi import Depends, FastAPI
from typing import Annotated

# Dependencias como funciones
async def get_db() -> AsyncGenerator[Session, None]:
  async with async_session() as session:
      yield session

async def get_cache() -> Redis:
  return redis_pool

# Servicio recibe dependencias por parámetro
class UsersService:
  def __init__(self, db: Session, cache: Redis):
      self.db = db
      self.cache = cache
  
  async def find_one(self, id: str) -> User | None:
      if cached := await self.cache.get(id):
          return cached
      return await self.db.get(User, id)

# Factory que compone las dependencias
async def get_users_service(
  db: Annotated[Session, Depends(get_db)],
  cache: Annotated[Redis, Depends(get_cache)],
) -> UsersService:
  return UsersService(db, cache)

# Endpoint
@app.get("/users/{user_id}")
async def get_user(
  user_id: str,
  service: Annotated[UsersService, Depends(get_users_service)],
) -> User:
  return await service.find_one(user_id)

Diferencias Clave

2. Anatomía de una Dependencia

Función Simple

from fastapi import Depends, Query
from typing import Annotated

# Dependencia básica
def get_pagination(
    page: int = Query(1, ge=1),
    size: int = Query(20, ge=1, le=100),
) -> dict:
    return {"skip": (page - 1) * size, "limit": size}

# Tipo reutilizable
Pagination = Annotated[dict, Depends(get_pagination)]

@app.get("/items")
async def list_items(pagination: Pagination) -> list[Item]:
    return await item_repo.list(**pagination)

Clase Callable

class PaginationParams:
    def __init__(
        self,
        page: int = Query(1, ge=1, description="Page number"),
        size: int = Query(20, ge=1, le=100, description="Items per page"),
    ):
        self.page = page
        self.size = size
        self.skip = (page - 1) * size
        self.limit = size

# El __init__ actúa como la función de dependencia
@app.get("/items")
async def list_items(
    pagination: Annotated[PaginationParams, Depends()],  # Depends() infiere la clase
) -> list[Item]:
    return await item_repo.list(skip=pagination.skip, limit=pagination.limit)

3. Yield Dependencies: Setup y Teardown

El patrón más potente de FastAPI es usar yield para manejar el ciclo de vida de recursos:

// NestJS: OnModuleInit / OnModuleDestroy
@Injectable()
export class DatabaseService implements OnModuleInit, OnModuleDestroy {
private pool: Pool;

async onModuleInit() {
  this.pool = await createPool(config);
}

async onModuleDestroy() {
  await this.pool.end();
}

getConnection() {
  return this.pool.connect();
}
}

# FastAPI: yield dependency
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession

async def get_db() -> AsyncGenerator[AsyncSession, None]:
  # SETUP: antes del yield
  async with async_session_maker() as session:
      try:
          yield session  # El endpoint recibe la sesión
          # COMMIT implícito si no hay errores
          await session.commit()
      except Exception:
          # ROLLBACK si hay error
          await session.rollback()
          raise
      # TEARDOWN: después del yield (siempre se ejecuta)
      # La sesión se cierra automáticamente por el context manager

Orden de Ejecución

async def dep_a():
    print("A: setup")
    yield "A"
    print("A: teardown")

async def dep_b(a: Annotated[str, Depends(dep_a)]):
    print("B: setup")
    yield f"B (got {a})"
    print("B: teardown")

@app.get("/test")
async def test_endpoint(b: Annotated[str, Depends(dep_b)]):
    print(f"Endpoint: {b}")
    return {"result": b}

# Output:
# A: setup
# B: setup
# Endpoint: B (got A)
# B: teardown  ← En orden inverso (LIFO)
# A: teardown

4. Scopes y Caching de Dependencias

Por defecto, FastAPI cachea las dependencias dentro del mismo request:

call_count = 0

async def get_settings() -> Settings:
    global call_count
    call_count += 1
    print(f"get_settings called: {call_count}")
    return Settings()

async def dep_a(settings: Annotated[Settings, Depends(get_settings)]):
    return f"A: {settings.app_name}"

async def dep_b(settings: Annotated[Settings, Depends(get_settings)]):
    return f"B: {settings.app_name}"

@app.get("/test")
async def test(
    a: Annotated[str, Depends(dep_a)],
    b: Annotated[str, Depends(dep_b)],
    settings: Annotated[Settings, Depends(get_settings)],
):
    # get_settings se llama SOLO UNA VEZ por request
    # Output: "get_settings called: 1"
    return {"a": a, "b": b}

Desactivar Cache (use_cache=False)

import uuid

def generate_request_id() -> str:
    return str(uuid.uuid4())

# Sin cache: genera un nuevo ID cada vez que se inyecta
@app.get("/test")
async def test(
    id1: Annotated[str, Depends(generate_request_id, use_cache=False)],
    id2: Annotated[str, Depends(generate_request_id, use_cache=False)],
):
    # id1 != id2
    return {"id1": id1, "id2": id2}

5. Dependency Overrides: Testing sin Mocks

Este es el killer feature para testing. No necesitas librerías de mocking externas:

# main.py
from fastapi import FastAPI, Depends
from typing import Annotated

app = FastAPI()

async def get_db() -> AsyncGenerator[Session, None]:
    async with production_session() as session:
        yield session

@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    db: Annotated[Session, Depends(get_db)],
) -> User:
    return await db.get(User, user_id)

# test_main.py
import pytest
from httpx import AsyncClient
from main import app, get_db

# Dependencia de test
async def get_test_db():
    async with test_session() as session:
        yield session

@pytest.fixture
def client():
    # Override la dependencia
    app.dependency_overrides[get_db] = get_test_db
    
    yield AsyncClient(app=app, base_url="http://test")
    
    # Limpiar después del test
    app.dependency_overrides.clear()

async def test_get_user(client):
    response = await client.get("/users/1")
    assert response.status_code == 200

Override con Factory

def get_mock_service(return_value: dict):
    """Factory que genera dependencias mock con valores específicos."""
    async def mock_service():
        return MockService(return_value)
    return mock_service

def test_specific_response():
    app.dependency_overrides[get_service] = get_mock_service({"id": 1, "name": "Test"})
    # ...

6. Composición Avanzada de Dependencias

Patrón: Repository + Service + Controller

from typing import Annotated, Protocol
from fastapi import Depends

# 1. Protocol (Interface)
class UserRepositoryProtocol(Protocol):
    async def get(self, id: int) -> User | None: ...
    async def create(self, data: UserCreate) -> User: ...

# 2. Implementación concreta
class SQLAlchemyUserRepository:
    def __init__(self, session: AsyncSession):
        self.session = session
    
    async def get(self, id: int) -> User | None:
        return await self.session.get(User, id)
    
    async def create(self, data: UserCreate) -> User:
        user = User(**data.model_dump())
        self.session.add(user)
        await self.session.flush()
        return user

# 3. Factory de Repository
async def get_user_repository(
    db: Annotated[AsyncSession, Depends(get_db)],
) -> UserRepositoryProtocol:
    return SQLAlchemyUserRepository(db)

# 4. Service (lógica de negocio)
class UserService:
    def __init__(self, repo: UserRepositoryProtocol, cache: Redis):
        self.repo = repo
        self.cache = cache
    
    async def get_user(self, id: int) -> User:
        # Cache-aside pattern
        cache_key = f"user:{id}"
        if cached := await self.cache.get(cache_key):
            return User.model_validate_json(cached)
        
        user = await self.repo.get(id)
        if not user:
            raise UserNotFoundError(id)
        
        await self.cache.set(cache_key, user.model_dump_json(), ex=300)
        return user

# 5. Factory de Service
async def get_user_service(
    repo: Annotated[UserRepositoryProtocol, Depends(get_user_repository)],
    cache: Annotated[Redis, Depends(get_redis)],
) -> UserService:
    return UserService(repo, cache)

# 6. Tipo reutilizable
UserServiceDep = Annotated[UserService, Depends(get_user_service)]

# 7. Endpoint limpio
@app.get("/users/{user_id}")
async def get_user(user_id: int, service: UserServiceDep) -> User:
    return await service.get_user(user_id)

Grafo de Dependencias Resultante

get_user (endpoint)
    └── get_user_service
            ├── get_user_repository
            │       └── get_db (yield)
            └── get_redis

7. Dependencias Globales y por Router

Dependencias Globales (Toda la App)

async def verify_api_key(x_api_key: str = Header(...)):
    if x_api_key != settings.API_KEY:
        raise HTTPException(status_code=403, detail="Invalid API key")

app = FastAPI(dependencies=[Depends(verify_api_key)])

# Todos los endpoints requieren API key
@app.get("/public")  # También requiere API key
async def public_endpoint():
    return {"message": "Hello"}

Dependencias por Router

from fastapi import APIRouter, Depends

# Router con dependencias propias
admin_router = APIRouter(
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(require_admin_role)],
)

@admin_router.get("/users")
async def list_all_users():  # require_admin_role se ejecuta automáticamente
    return await user_repo.list_all()

@admin_router.delete("/users/{user_id}")
async def delete_user(user_id: int):  # require_admin_role se ejecuta automáticamente
    return await user_repo.delete(user_id)

# Incluir en la app
app.include_router(admin_router)

8. Contextual Dependencies con contextvars

Para compartir estado a través del request sin pasarlo explícitamente:

from contextvars import ContextVar
from fastapi import Request
from typing import Annotated

# Context variable para el request ID
request_id_ctx: ContextVar[str] = ContextVar("request_id", default="")

async def inject_request_id(request: Request):
    """Middleware que inyecta el request ID en el contexto."""
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    request_id_ctx.set(request_id)
    return request_id

RequestId = Annotated[str, Depends(inject_request_id)]

# En cualquier parte del código (sin inyección explícita)
def get_current_request_id() -> str:
    return request_id_ctx.get()

# Service que usa el context
class AuditService:
    async def log_action(self, action: str):
        request_id = get_current_request_id()  # Acceso sin Depends
        await self.repo.create(AuditLog(
            request_id=request_id,
            action=action,
            timestamp=datetime.utcnow(),
        ))

9. Lifespan Events: Startup y Shutdown

Para recursos que viven toda la vida de la aplicación:

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # STARTUP
    print("Starting up...")
    app.state.db_pool = await create_db_pool()
    app.state.redis = await create_redis_pool()
    app.state.http_client = httpx.AsyncClient()
    
    yield  # La aplicación corre aquí
    
    # SHUTDOWN
    print("Shutting down...")
    await app.state.db_pool.close()
    await app.state.redis.close()
    await app.state.http_client.aclose()

app = FastAPI(lifespan=lifespan)

# Acceder al state en dependencias
async def get_db_pool(request: Request) -> Pool:
    return request.app.state.db_pool

10. Anti-patterns y Best Practices

❌ Anti-pattern: Lógica de negocio en dependencias

# MAL: La dependencia hace demasiado
async def get_user_with_permissions(
    user_id: int,
    db: Session = Depends(get_db),
):
    user = await db.get(User, user_id)
    if not user:
        raise HTTPException(404)  # ❌ HTTP en dependencia
    
    permissions = await db.execute(...)  # ❌ Lógica de negocio
    user.permissions = permissions
    return user

# BIEN: Dependencia solo resuelve recursos
async def get_db() -> AsyncGenerator[Session, None]:
    async with session_maker() as session:
        yield session

# La lógica va en el Service
class UserService:
    async def get_with_permissions(self, user_id: int) -> User:
        user = await self.repo.get(user_id)
        if not user:
            raise UserNotFoundError(user_id)  # ✅ Excepción de dominio
        return await self.enrich_with_permissions(user)

❌ Anti-pattern: Dependencias circulares

# MAL: A depende de B, B depende de A
async def get_a(b = Depends(get_b)):
    return A(b)

async def get_b(a = Depends(get_a)):  # ❌ Circular!
    return B(a)

# BIEN: Usar Protocol para romper el ciclo
class AProtocol(Protocol):
    def do_something(self) -> str: ...

async def get_b() -> B:
    return B()

async def get_a(b: Annotated[B, Depends(get_b)]) -> A:
    return A(b)  # A usa B, no al revés

11. Tabla de Patrones de DI

Conclusión

El sistema de DI de FastAPI es más simple pero igualmente potente que NestJS:

1. Funciones en lugar de clases — Más Pythonic y fácil de componer
2. yield para lifecycle — Setup/teardown sin decoradores especiales
3. Caching automático — Misma instancia dentro del request
4. Override nativo — Testing sin frameworks de mocking

Mental model Senior: Piensa en las dependencias como un grafo de funciones que se resuelve lazy. Cada nodo es una función, cada arista es un Depends().

En el siguiente capítulo, veremos cómo manejar errores y crear middleware personalizado.