Inyección de dependencias en modelos Pydantic mediante fábricas y metadatos

La gestión de dependencias y la validación de datos suelen tratarse como preocupaciones separadas en el desarrollo con Python. Sin embargo, Pydantic permite unificar ambos aspectos utilizando su sistema de definición de campos. Al aprovechar los metadatos y las fábricas de valores por defecto, es posible inyectar servicios, configuraciones o conexiones de manera declarativa, eliminando la necesidad de instanciar dependencias de forma rígida dentro de la lógica de negocio.

Arquitectura de metadatos en Pydantic

La capacidad de inyectar depandencias radica en la estructura interna de FieldInfo. Cuando definimos un campo, Pydantic encapsula sus restricciones, alias y comportamientos por defecto en este objeto. Esto nos permite adjuntar lógica de resolución directamente en la firma del modelo, separando la declaración de la implementación.

from pydantic.fields import FieldInfo
from typing import Any, Callable

class InjectableField(FieldInfo):
    """Extensión de FieldInfo para encapsular proveedores de dependencias."""
    def __init__(self, provider: Callable, **kwargs: Any):
        self.provider = provider
        # Delegamos la creación a la fábrica por defecto nativa
        super().__init__(default_factory=provider, **kwargs)

Esta abstracción demuestra cómo los metadatos de los campos actúan como el punto de anclaje para interceptar y proveer recursos externos.

Inyección mediante fábricas por defecto

El parámetro default_factory es el mecanismo más directo para la inyección. En lugar de asignar un valor estático, delegamos la creación del objeto a una función externa. Esto es ideal para recursos que deben instanciarse bajo demanda y que no son serializables.

from pydantic import BaseModel, Field
import uuid

class LoggerService:
    def log(self, msg: str) -> None:
        pass

def create_logger() -> LoggerService:
    return LoggerService()

class AuditRecord(BaseModel):
    trace_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    # Se inyecta el servicio y se excluye de la serialización JSON
    logger: LoggerService = Field(default_factory=create_logger, exclude=True)
    action: str

En este escenario, cada vez que se crea un AuditRecord, se genera un identificador único y se provisiona una instancia del servicio de registro, manteniendo el modelo limpio de lógicas de inicialización complejas.

Uso de Annotated para tipos reutilizables

Para evitar repetir la configuración de inyección en múltiples modelos, las versiones modernas de Python permiten empaquetar la lógica dentro de un alias de tipo utilizando typing.Annotated.

from typing import Annotated
from pydantic import BaseModel, Field

class RedisClient:
    def get_connection(self) -> str:
        return "redis_conn_pool"

def resolve_redis() -> RedisClient:
    return RedisClient()

# Definición del tipo con su dependencia inyectada
CacheDependency = Annotated[RedisClient, Field(default_factory=resolve_redis, exclude=True)]

class SessionManager(BaseModel):
    cache: CacheDependency
    session_id: str

Cualquier modelo que utilice CacheDependency recibirá automáticamente la conexión configurada, centralizando la gestión de recursso compartidos.

Resolución dinámica con validadores

Cuando una dependencia requiere datos de otros campos para ser resuelta, los validadores de modelo ofrecen un interceptor ideal. Podemos alterar o completar el estado del objeto antes de que la validación finalice.

from pydantic import BaseModel, Field, model_validator
from typing import Optional

class PermissionsService:
    def get_roles(self, uid: int) -> list[str]:
        return ["admin"] if uid == 1 else ["user"]

class UserContext(BaseModel):
    user_id: int
    roles: Optional[list[str]] = None
    permissions_svc: PermissionsService = Field(default_factory=PermissionsService, exclude=True)

    @model_validator(mode='after')
    def inject_roles(self) -> 'UserContext':
        if self.roles is None:
            self.roles = self.permissions_svc.get_roles(self.user_id)
        return self

Este patrón permite que el modelo consulte servicios externos basándose en su propio estado inicial, actuando como un agregador de datos inteligente.

Gestión de configuraciones de entorno

Un caso de uso habitual es la lectura de variables de entorno sin acoplar el modelo al módulo del sistema operativo de forma explícita en cada atributo.

import os
from pydantic import BaseModel, Field

class EnvProvider:
    @staticmethod
    def get(key: str, cast_type: type = str):
        val = os.getenv(key)
        if val is None:
            raise EnvironmentError(f"Missing env var: {key}")
        return cast_type(val)

class SystemSettings(BaseModel):
    db_uri: str = Field(default_factory=lambda: EnvProvider.get("DB_URI"))
    max_retries: int = Field(default_factory=lambda: EnvProvider.get("RETRIES", int))
    enable_metrics: bool = Field(default_factory=lambda: EnvProvider.get("METRICS") == "1")

Al utilizar una clase proveedora junto con default_factory, la lectura del entorno se evalúa de forma diferida y el modelo permanece agnóstico respecto al origen de los datos.

Beneficios arquitectónicos

  • Desacoplamiento: Los modelos de datos no necesitan conocer los detalles de instanciación de los servicios externos.
  • Pruebas unitarias: Las fábricas pueden ser fácilmente sobrescritas o simuladas (mocked) inyectando funciones alternativas durante la ejecución de tests.
  • Evaluación diferida: Las dependencias pesadas, como conexiones a bases de datos o clientes HTTP, solo se crean cuando el modelo es instanciado, optimizando el uso de memoria.
  • Limpieza en serialización: El uso de exclude=True en campos inyectados asegura que los servicios no se filtren accidentalmente en respuestas JSON.

Etiquetas: pydantic Python dependency-injection type-hints data-validation

Publicado el 8-2 12:05