Concepto clave

La inyección de dependencias es un patrón de diseño que permite desacoplar componentes de software al pasar sus dependencias desde el exterior, en lugar de crearlas internamente. En FastAPI, esto se implementa mediante el sistema de dependencias integrado, que gestiona automáticamente la creación y limpieza de recursos.

Imagina un restaurante donde cada cocinero compra sus propios ingredientes directamente del mercado. Esto es ineficiente y duplica esfuerzos. Con inyección de dependencias, tendrías un proveedor central que entrega los ingredientes a cada cocinero cuando los necesita. En desarrollo backend, los "ingredientes" son servicios como conexiones a bases de datos, clientes de API externas, o lógica de negocio reutilizable.

Este patrón mejora la testabilidad (puedes mockear dependencias fácilmente), la mantenibilidad (cambios en un servicio no afectan a todos los componentes) y la reutilización (un mismo servicio puede usarse en múltiples endpoints). FastAPI lo hace especialmente elegante con su sistema declarativo.

Cómo funciona en la práctica

FastAPI implementa la inyección de dependencias mediante funciones o clases que se declaran como parámetros en los endpoints. El framework se encarga de ejecutarlas y pasar sus resultados automáticamente. Veamos el proceso paso a paso:

Definición de la dependencia: Creas una función (o clase) que devuelve el servicio o recurso necesario. Puede incluir lógica de inicialización, validación o limpieza.
Declaración en el endpoint: Especificas la dependencia como parámetro usando Depends() de FastAPI.
Ejecución automática: FastAPI ejecuta la dependencia antes del endpoint y pasa su resultado como argumento.
Gestión del ciclo de vida: Para recursos como conexiones a BD, puedes usar yield para limpieza automática después de la respuesta.

Este enfoque permite compartir lógica común (como autenticación, logging o acceso a datos) entre múltiples endpoints sin duplicar código.

Código en acción

Veamos un ejemplo completo que muestra el "antes y después" de refactorizar un servicio de base de datos:

Antes: Sin inyección de dependencias

from fastapi import FastAPI, HTTPException
import asyncpg

app = FastAPI()

# Conexión creada directamente en cada endpoint (problema: duplicación)
async def get_db_connection():
    return await asyncpg.connect("postgresql://user:pass@localhost/db")

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    conn = await get_db_connection()  # Conexión creada aquí
    try:
        user = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
        if not user:
            raise HTTPException(status_code=404, detail="User not found")
        return dict(user)
    finally:
        await conn.close()  # Limpieza manual

@app.post("/users/")
async def create_user(name: str, email: str):
    conn = await get_db_connection()  # Duplicación de código
    try:
        await conn.execute("INSERT INTO users(name, email) VALUES($1, $2)", name, email)
        return {"message": "User created"}
    finally:
        await conn.close()

Después: Con inyección de dependencias

from fastapi import FastAPI, HTTPException, Depends
from contextlib import asynccontextmanager
import asyncpg

app = FastAPI()

# Dependencia reutilizable con gestión automática de ciclo de vida
@asynccontextmanager
async def get_db():
    conn = await asyncpg.connect("postgresql://user:pass@localhost/db")
    try:
        yield conn  # Entrega la conexión al endpoint
    finally:
        await conn.close()  # Limpieza automática después del yield

# Función de dependencia para FastAPI
async def db_dependency():
    async with get_db() as conn:
        yield conn

@app.get("/users/{user_id}")
async def get_user(user_id: int, conn = Depends(db_dependency)):
    user = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return dict(user)

@app.post("/users/")
async def create_user(name: str, email: str, conn = Depends(db_dependency)):
    await conn.execute("INSERT INTO users(name, email) VALUES($1, $2)", name, email)
    return {"message": "User created"}

# Ejemplo con dependencia parametrizada (más avanzado)
async def get_user_service(conn = Depends(db_dependency)):
    # Servicio reutilizable que encapsula lógica de negocio
    class UserService:
        def __init__(self, connection):
            self.conn = connection
        
        async def get_by_id(self, user_id: int):
            return await self.conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
        
        async def create(self, name: str, email: str):
            return await self.conn.execute("INSERT INTO users(name, email) VALUES($1, $2)", name, email)
    
    return UserService(conn)

@app.get("/users/v2/{user_id}")
async def get_user_v2(user_id: int, user_service = Depends(get_user_service)):
    user = await user_service.get_by_id(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return dict(user)

Errores comunes

Crear dependencias con estado mutable compartido: Si una dependencia devuelve un objeto mutable (como una lista o diccionario) y múltiples endpoints lo modifican, pueden producirse condiciones de carrera. Solución: Usa yield para crear instancias frescas o diseña servicios sin estado.
No gestionar el ciclo de vida de recursos: Olvidar cerrar conexiones a BD o archivos puede causar fugas de memoria. Usa @asynccontextmanager con yield para limpieza automática.
Abusar de dependencias para lógica compleja: Las dependencias deben ser simples y enfocadas en proporcionar recursos. Si incluyen lógica de negocio pesada, dificultan las pruebas. Extrae la lógica a servicios separados.
Ignorar el tipado: No definir tipos en las dependencias reduce la claridad y el soporte del IDE. Siempre usa anotaciones de tipo para mejorar el mantenimiento.
Duplicar dependencias similares: Crear múltiples dependencias para variaciones menores del mismo servicio. Mejor: parametriza una dependencia base usando funciones que retornen otras funciones.

Checklist de dominio

Puedo explicar la diferencia entre inyección de dependencias y creación directa de instancias.
He implementado al menos una dependencia con yield para gestión automática de recursos.
Puedo crear dependencias parametrizadas que acepten argumentos dinámicos.
He refactorizado código duplicado en al menos un proyecto real usando dependencias.
Puedo mockear dependencias en pruebas unitarias sin modificar el código de producción.
Entiendo cuándo usar clases vs. funciones como dependencias y sus implicaciones.
He utilizado dependencias para cross-cutting concerns como autenticación, logging o validación.

Refactorización de un servicio de notificaciones con inyección de dependencias

En este ejercicio, refactorizarás un módulo de notificaciones que actualmente crea sus dependencias directamente, aplicando inyección de dependencias para hacerlo más testeable y mantenible.

Contexto inicial: Tienes un servicio de notificaciones que envía emails y mensajes SMS. Actualmente, los clientes de email y SMS se instancian directamente dentro de las funciones, lo que dificulta las pruebas y la reutilización.

Código base:

# servicio_notificaciones.py (versión actual)
import smtplib
from twilio.rest import Client

class NotificacionService:
    def enviar_email(self, destinatario: str, mensaje: str):
        # Configuración hardcodeada (problema)
        server = smtplib.SMTP("smtp.gmail.com", 587)
        server.starttls()
        server.login("[email protected]", "password123")
        server.sendmail("[email protected]", destinatario, mensaje)
        server.quit()
    
    def enviar_sms(self, numero: str, mensaje: str):
        # Credenciales hardcodeadas (problema)
        client = Client("AC123", "token456")
        client.messages.create(body=mensaje, from_="+1234567890", to=numero)

Paso 1: Extraer dependencias: Crea dos funciones de dependencia: get_email_client() y get_sms_client(). Deben recibir configuración desde variables de entorno (usa os.getenv) en lugar de valores hardcodeados.
Paso 2: Refactorizar el servicio: Modifica NotificacionService para recibir los clientes como parámetros en su constructor (inyección por constructor).
Paso 3: Integrar con FastAPI: Crea un endpoint POST /notificar que use Depends() para inyectar una instancia de NotificacionService preconfigurada.
Paso 4: Configuración flexible: Añade la posibilidad de desactivar SMS o email mediante configuración, haciendo que las dependencias devuelvan None o un mock en esos casos.
Paso 5: Prueba la solución: Escribe una prueba unitaria que mockee los clientes para verificar que el servicio funciona sin conexiones reales.

Entrega: Un archivo Python con el servicio refactorizado y el endpoint FastAPI funcionando.

Pistas

Usa el patrón de inyección por constructor para hacer el servicio más testeable.
Considera usar clases como dependencias para encapsular la lógica de configuración.
Para el paso 4, puedes crear una dependencia que lea una variable de entorno como 'NOTIFICACIONES_SMS_HABILITADO'.

Evalua tu comprension

Completa el quiz interactivo de arriba para ganar XP.

Implementar Inyección de Dependencias para Servicios Reutilizables