Módulo 6: Cloud Migration Patterns

4. Dependency Injection para boto3 Clients

Descripción

En esta cápsula vas a implementar dependency injection para boto3 clients. En lugar de que tu lógica de negocio construya sus propios clientes S3 o Lambda (decidiendo endpoints, credenciales, región), va a recibirlos ya configurados. Un factory pattern crea los clients según el entorno, y tu código de negocio simplemente los usa. Es la diferencia entre una función que sabe demasiado sobre infraestructura y una función que solo sabe procesar documentos.

Contexto: En la cápsula 02 abstraíste el entorno. En la 03 construiste config management tipado. Ahora vas a conectar ambos: el factory lee la configuración, construye clients boto3 correctamente configurados, y los inyecta donde se necesiten. Tu DocumentProcessor no va a importar boto3, no va a leer variables de entorno, no va a decidir si usa LocalStack o AWS. Recibe un client y trabaja. Si mañana cambias de S3 a MinIO, solo cambias el factory — el processor ni se entera.

Por Qué Dependency Injection para Cloud Clients

El anti-patrón: lógica de negocio que construye clients

# ❌ Anti-patrón — la lógica de negocio sabe demasiado
import boto3
import os


class DocumentProcessor:
    def __init__(self):
        endpoint = os.environ.get("AWS_ENDPOINT_URL")
        self.s3 = boto3.client(
            "s3",
            endpoint_url=endpoint,
            aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID"),
            aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY"),
            region_name=os.environ.get("AWS_REGION", "us-east-1"),
        )
        self.bucket = os.environ.get("S3_BUCKET", "default")

    def process(self, key: str) -> dict:
        response = self.s3.get_object(Bucket=self.bucket, Key=key)
        # ... procesamiento ...

Problemas:

❌ DocumentProcessor conoce detalles de infraestructura (endpoint_url, credenciales)
❌ Imposible de testear con un mock S3 sin manipular variables de entorno
❌ Cada clase que necesite S3 repite la misma construcción de client
❌ Cambiar de proveedor (S3 → MinIO → GCS) requiere editar lógica de negocio

El patrón: inyectar dependencias configuradas

# ✅ Patrón correcto — la lógica de negocio es pura
class DocumentProcessor:
    def __init__(self, s3_client, bucket_name: str):
        self.s3 = s3_client
        self.bucket = bucket_name

    def process(self, key: str) -> dict:
        response = self.s3.get_object(Bucket=self.bucket, Key=key)
        # ... procesamiento ...

DocumentProcessor no sabe qué es boto3. No sabe si el client habla con LocalStack o AWS. No importa os. Recibe un client que tiene .get_object() y un bucket name — y trabaja.

Beneficios concretos

Sin DI                               Con DI
──────────────────────────────       ──────────────────────────────
Clase crea su client                  Clase recibe client
  → acoplada a boto3                    → funciona con cualquier client
  → lee env vars                        → no depende de env vars
  → difícil de testear                  → fácil de testear (mock)
  → cambiar proveedor = reescribir     → cambiar proveedor = cambiar factory
  → N clases × construcción repetida   → 1 factory × N clases

El Factory Pattern para boto3

ClientFactory: creación centralizada de clients

"""clients/factory.py — Factory para crear boto3 clients."""

import boto3
from typing import Any
from config.settings import Settings


class ClientFactory:
    """Crea boto3 clients configurados para el entorno actual.

    Centraliza la lógica de construcción: endpoints, credenciales,
    región. Ningún otro módulo necesita saber cómo se configura boto3.
    """

    def __init__(self, settings: Settings):
        self.settings = settings
        self._base_kwargs = self._build_base_kwargs()
        self._clients: dict[str, Any] = {}

    def _build_base_kwargs(self) -> dict:
        """Construye kwargs comunes para todos los clients."""
        kwargs = {"region_name": self.settings.aws_region}

        if self.settings.aws_endpoint_url:
            kwargs["endpoint_url"] = self.settings.aws_endpoint_url

        if self.settings.aws_access_key_id:
            kwargs["aws_access_key_id"] = self.settings.aws_access_key_id
            kwargs["aws_secret_access_key"] = self.settings.aws_secret_access_key

        return kwargs

    def get_client(self, service: str) -> Any:
        """Retorna un client boto3 para el servicio indicado.

        Los clients se cachean: el mismo servicio retorna la misma instancia.
        """
        if service not in self._clients:
            self._clients[service] = boto3.client(service, **self._base_kwargs)
        return self._clients[service]

    @property
    def s3(self) -> Any:
        return self.get_client("s3")

    @property
    def lambda_client(self) -> Any:
        return self.get_client("lambda")

    @property
    def logs(self) -> Any:
        return self.get_client("logs")

    @property
    def secretsmanager(self) -> Any:
        return self.get_client("secretsmanager")

    def health_check(self) -> dict:
        """Verifica conectividad de todos los clients cacheados."""
        results = {}
        for service, client in self._clients.items():
            try:
                if service == "s3":
                    client.list_buckets()
                elif service == "lambda":
                    client.list_functions(MaxItems=1)
                elif service == "logs":
                    client.describe_log_groups(limit=1)
                results[service] = "healthy"
            except Exception as e:
                results[service] = f"unhealthy: {e}"
        return results

    def __repr__(self) -> str:
        return (
            f"ClientFactory(env={self.settings.environment}, "
            f"endpoint={self.settings.aws_endpoint_url or 'AWS'}, "
            f"cached_clients={list(self._clients.keys())})"
        )

Uso del factory

from config.loader import get_settings
from clients.factory import ClientFactory

settings = get_settings()
factory = ClientFactory(settings)

# Los clients están configurados para el entorno correcto
s3 = factory.s3
s3.list_buckets()  # Habla con LocalStack o AWS según ENVIRONMENT

lambda_client = factory.lambda_client
lambda_client.list_functions()

print(factory)
# ClientFactory(env=local, endpoint=http://localhost:4566, cached_clients=['s3', 'lambda'])

Inyectando Clients en Servicios

DocumentProcessor con DI

"""services/document_processor.py — Procesador de documentos con DI."""

import json
from datetime import datetime
from typing import Any


class DocumentProcessor:
    """Procesa documentos AI.

    No conoce boto3, no conoce el entorno, no lee variables de entorno.
    Recibe clients y configuración — y trabaja.
    """

    def __init__(self, s3_client: Any, bucket: str):
        self.s3 = s3_client
        self.bucket = bucket

    def get_prompt(self, name: str, version: str) -> str:
        key = f"prompts/{name}/{version}/system.txt"
        response = self.s3.get_object(Bucket=self.bucket, Key=key)
        return response["Body"].read().decode("utf-8")

    def store_document(self, collection: str, doc_id: str, data: dict) -> str:
        key = f"documents/{collection}/{doc_id}.json"
        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=json.dumps(data, ensure_ascii=False).encode("utf-8"),
            ContentType="application/json",
        )
        return key

    def save_response(self, request_id: str, result: dict) -> str:
        now = datetime.utcnow()
        key = f"responses/{now.strftime('%Y/%m/%d')}/{request_id}.json"
        payload = {"request_id": request_id, "timestamp": now.isoformat(), **result}
        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
            ContentType="application/json",
        )
        return key

InferenceService con DI

"""services/inference.py — Servicio de inferencia con DI."""

import json
from typing import Any, Optional


class InferenceService:
    """Ejecuta inferencia invocando una Lambda function.

    Recibe un lambda client y el nombre de la función.
    No sabe si la Lambda corre en LocalStack o AWS.
    """

    def __init__(self, lambda_client: Any, function_name: str):
        self.lambda_client = lambda_client
        self.function_name = function_name

    def invoke(self, payload: dict) -> dict:
        response = self.lambda_client.invoke(
            FunctionName=self.function_name,
            InvocationType="RequestResponse",
            Payload=json.dumps(payload).encode("utf-8"),
        )

        status_code = response["StatusCode"]
        response_payload = json.loads(
            response["Payload"].read().decode("utf-8")
        )

        if status_code != 200:
            raise RuntimeError(
                f"Lambda retornó status {status_code}: {response_payload}"
            )

        return response_payload

    def invoke_async(self, payload: dict) -> str:
        response = self.lambda_client.invoke(
            FunctionName=self.function_name,
            InvocationType="Event",
            Payload=json.dumps(payload).encode("utf-8"),
        )
        return response["StatusCode"]

Wiring: conectar factory con servicios

"""app.py — Wiring: conecta config, factory, y servicios."""

from config.loader import get_settings
from clients.factory import ClientFactory
from services.document_processor import DocumentProcessor
from services.inference import InferenceService


def create_app():
    """Construye la aplicación con todas las dependencias inyectadas."""
    settings = get_settings()
    factory = ClientFactory(settings)

    processor = DocumentProcessor(
        s3_client=factory.s3,
        bucket=settings.s3_bucket,
    )

    inference = InferenceService(
        lambda_client=factory.lambda_client,
        function_name=settings.lambda_function_name,
    )

    return {
        "settings": settings,
        "factory": factory,
        "processor": processor,
        "inference": inference,
    }


# Uso
app = create_app()
print(f"Entorno: {app['settings'].environment}")
print(f"Factory: {app['factory']}")

prompt = app["processor"].get_prompt("summarizer", "v1")
print(f"Prompt cargado: {prompt[:60]}...")

Cambia ENVIRONMENT=local a ENVIRONMENT=staging y el mismo create_app() construye todo contra AWS. Zero cambios en DocumentProcessor, InferenceService, o cualquier otro servicio.

Service Container: Organización Avanzada

Container que administra todos los servicios

"""services/container.py — Service container con DI."""

from dataclasses import dataclass
from typing import Any
from config.settings import Settings
from clients.factory import ClientFactory
from services.document_processor import DocumentProcessor
from services.inference import InferenceService


@dataclass
class ServiceContainer:
    """Contiene todos los servicios de la aplicación, ya inyectados."""
    settings: Settings
    factory: ClientFactory
    processor: DocumentProcessor
    inference: InferenceService

    @classmethod
    def create(cls, settings: Settings | None = None) -> "ServiceContainer":
        """Factory method que construye el container completo."""
        if settings is None:
            from config.loader import get_settings
            settings = get_settings()

        factory = ClientFactory(settings)

        processor = DocumentProcessor(
            s3_client=factory.s3,
            bucket=settings.s3_bucket,
        )

        inference = InferenceService(
            lambda_client=factory.lambda_client,
            function_name=settings.lambda_function_name,
        )

        return cls(
            settings=settings,
            factory=factory,
            processor=processor,
            inference=inference,
        )

    def health(self) -> dict:
        """Health check de todos los servicios."""
        return {
            "environment": self.settings.environment,
            "clients": self.factory.health_check(),
        }

Uso en Lambda handler

"""handler.py — Lambda handler con DI via ServiceContainer."""

import json
from services.container import ServiceContainer

container: ServiceContainer | None = None


def get_container() -> ServiceContainer:
    """Lazy initialization del container (reutilizado entre invocaciones)."""
    global container
    if container is None:
        container = ServiceContainer.create()
    return container


def lambda_handler(event, context):
    """Handler de Lambda con dependencias inyectadas."""
    c = get_container()

    path = event.get("rawPath", event.get("path", ""))
    method = event.get("requestContext", {}).get("http", {}).get("method", "GET")

    if path == "/health" and method == "GET":
        health = c.health()
        return {
            "statusCode": 200,
            "body": json.dumps(health),
        }

    if path == "/process" and method == "POST":
        body = json.loads(event.get("body", "{}"))
        prompt_name = body.get("prompt_name", "summarizer")
        prompt_version = body.get("prompt_version", "v1")
        document = body.get("document", {})

        prompt = c.processor.get_prompt(prompt_name, prompt_version)
        stored_key = c.processor.store_document(
            "inbox", document.get("id", "unknown"), document
        )
        result = {
            "prompt_used": f"{prompt_name}/{prompt_version}",
            "document_stored": stored_key,
            "processed": True,
        }
        response_key = c.processor.save_response(
            f"req-{context.aws_request_id[:8] if context else 'local'}",
            result,
        )
        result["response_key"] = response_key

        return {
            "statusCode": 200,
            "body": json.dumps(result),
        }

    return {
        "statusCode": 404,
        "body": json.dumps({"error": "Not found"}),
    }

Testing con DI: La Ventaja Real

Mock clients para tests unitarios

La ventaja más práctica de DI: testear sin LocalStack ni AWS.

"""tests/test_document_processor.py — Tests con mock clients."""

import json
from io import BytesIO
from unittest.mock import MagicMock
from services.document_processor import DocumentProcessor


def make_s3_response(content: str) -> dict:
    """Helper: crea un response de S3 mockeado."""
    body = MagicMock()
    body.read.return_value = content.encode("utf-8")
    return {"Body": body}


def test_get_prompt():
    mock_s3 = MagicMock()
    mock_s3.get_object.return_value = make_s3_response(
        "Eres un asistente de resúmenes."
    )

    processor = DocumentProcessor(s3_client=mock_s3, bucket="test-bucket")
    result = processor.get_prompt("summarizer", "v1")

    assert result == "Eres un asistente de resúmenes."
    mock_s3.get_object.assert_called_once_with(
        Bucket="test-bucket",
        Key="prompts/summarizer/v1/system.txt",
    )


def test_store_document():
    mock_s3 = MagicMock()

    processor = DocumentProcessor(s3_client=mock_s3, bucket="test-bucket")
    key = processor.store_document(
        "knowledge-base",
        "doc-001",
        {"title": "Test", "content": "Contenido de prueba"},
    )

    assert key == "documents/knowledge-base/doc-001.json"
    mock_s3.put_object.assert_called_once()
    call_kwargs = mock_s3.put_object.call_args[1]
    assert call_kwargs["Bucket"] == "test-bucket"
    assert call_kwargs["Key"] == "documents/knowledge-base/doc-001.json"

    body = json.loads(call_kwargs["Body"].decode("utf-8"))
    assert body["title"] == "Test"


def test_save_response():
    mock_s3 = MagicMock()

    processor = DocumentProcessor(s3_client=mock_s3, bucket="test-bucket")
    key = processor.save_response("req-abc123", {"status": "ok"})

    assert key.startswith("responses/")
    assert "req-abc123" in key
    mock_s3.put_object.assert_called_once()

Estos tests corren en millisegundos, sin Docker, sin LocalStack, sin AWS. El MagicMock reemplaza al client boto3, y DocumentProcessor funciona idéntico porque no le importa quién es el client — solo que tenga .get_object() y .put_object().

Tests de integración con factory real

"""tests/test_integration.py — Tests de integración con factory real."""

import pytest
from config.settings import Settings, EnvironmentName
from clients.factory import ClientFactory
from services.document_processor import DocumentProcessor


@pytest.fixture
def local_settings():
    return Settings(
        environment=EnvironmentName.LOCAL,
        aws_endpoint_url="http://localhost:4566",
        aws_access_key_id="test",
        aws_secret_access_key="test",
        s3_bucket="test-integration-bucket",
    )


@pytest.fixture
def processor(local_settings):
    factory = ClientFactory(local_settings)
    s3 = factory.s3
    try:
        s3.create_bucket(Bucket=local_settings.s3_bucket)
    except s3.exceptions.BucketAlreadyOwnedByYou:
        pass

    return DocumentProcessor(s3_client=s3, bucket=local_settings.s3_bucket)


def test_roundtrip_prompt(processor):
    """Sube y recupera un prompt template — integración real."""
    processor.s3.put_object(
        Bucket=processor.bucket,
        Key="prompts/test/v1/system.txt",
        Body=b"Test prompt content",
    )
    result = processor.get_prompt("test", "v1")
    assert result == "Test prompt content"


def test_roundtrip_document(processor):
    """Sube y verifica un documento — integración real."""
    key = processor.store_document(
        "test-collection",
        "doc-integration",
        {"title": "Integration Test", "status": "active"},
    )

    response = processor.s3.get_object(Bucket=processor.bucket, Key=key)
    import json
    data = json.loads(response["Body"].read().decode("utf-8"))
    assert data["title"] == "Integration Test"

Patrón Avanzado: Client con Retry Built-in

Wrapper que agrega retry a cualquier client

"""clients/resilient.py — Client wrapper con retry automático."""

import time
import logging
from typing import Any

logger = logging.getLogger(__name__)


class ResilientClient:
    """Wrapper que agrega retry automático a un boto3 client.

    No modifica el client original — envuelve llamadas con retry logic.
    """

    def __init__(
        self,
        client: Any,
        max_retries: int = 3,
        base_delay: float = 1.0,
        service_name: str = "unknown",
    ):
        self._client = client
        self._max_retries = max_retries
        self._base_delay = base_delay
        self._service_name = service_name

    def __getattr__(self, name: str):
        attr = getattr(self._client, name)
        if not callable(attr):
            return attr

        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(1, self._max_retries + 1):
                try:
                    return attr(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    if attempt < self._max_retries:
                        delay = self._base_delay * (2 ** (attempt - 1))
                        logger.warning(
                            f"[{self._service_name}] {name} falló (intento {attempt}/"
                            f"{self._max_retries}): {e}. Retry en {delay}s"
                        )
                        time.sleep(delay)
                    else:
                        logger.error(
                            f"[{self._service_name}] {name} falló tras "
                            f"{self._max_retries} intentos: {e}"
                        )
            raise last_exception

        return wrapper

Factory con clients resilientes

"""clients/factory.py — Extensión con soporte para resilient clients."""

from clients.resilient import ResilientClient


class ClientFactory:
    # ... (código anterior) ...

    def get_resilient_client(self, service: str) -> ResilientClient:
        """Retorna un client con retry automático."""
        base_client = self.get_client(service)
        return ResilientClient(
            client=base_client,
            max_retries=self.settings.max_retries,
            base_delay=1.0,
            service_name=service,
        )

    @property
    def resilient_s3(self) -> ResilientClient:
        return self.get_resilient_client("s3")

Uso:

factory = ClientFactory(settings)
s3 = factory.resilient_s3

# Esto reintenta automáticamente si S3 falla temporalmente
s3.get_object(Bucket="my-bucket", Key="my-key")

Troubleshooting

Problema 1: "AttributeError: 'MagicMock' has no attribute 'exceptions'"

Al mockear S3 para tests, s3.exceptions.ClientError falla porque MagicMock no tiene exceptions.

from botocore.exceptions import ClientError

# En vez de mock_s3.exceptions.ClientError, importa directamente:
mock_s3.head_bucket.side_effect = ClientError(
    {"Error": {"Code": "404", "Message": "Not Found"}},
    "HeadBucket",
)

Problema 2: El factory cachea un client que luego falla

Si LocalStack se reinicia, el client cacheado pierde conexión.

class ClientFactory:
    def reset_client(self, service: str):
        """Invalida el cache de un client específico."""
        self._clients.pop(service, None)

    def reset_all(self):
        """Invalida todos los clients cacheados."""
        self._clients.clear()

Problema 3: Circular import entre settings y factory

settings.py importa de factory.py que importa de settings.py.

# Solución: factory importa Settings, no el loader
# factory.py
from config.settings import Settings  # ← clase, no loader

# app.py (wiring)
from config.loader import get_settings
from clients.factory import ClientFactory

settings = get_settings()
factory = ClientFactory(settings)

Problema 4: Lambda handler necesita el container pero no puede importar config

El handler debe funcionar como punto de entrada sin dependencias circulares.

# handler.py — lazy initialization
container = None

def get_container():
    global container
    if container is None:
        from services.container import ServiceContainer
        container = ServiceContainer.create()
    return container

Ejercicios Prácticos

Ejercicio 1: Factory con SageMaker condicional

Extiende el ClientFactory para que solo cree el client de SageMaker si feature_sagemaker_enabled=True en settings. Si se solicita el client de SageMaker cuando no está habilitado, lanza una excepción descriptiva.

Ver solución

import boto3
from typing import Any
from config.settings import Settings


class SageMakerDisabledError(Exception):
    pass


class ClientFactory:
    def __init__(self, settings: Settings):
        self.settings = settings
        self._base_kwargs = self._build_base_kwargs()
        self._clients: dict[str, Any] = {}

    def _build_base_kwargs(self) -> dict:
        kwargs = {"region_name": self.settings.aws_region}
        if self.settings.aws_endpoint_url:
            kwargs["endpoint_url"] = self.settings.aws_endpoint_url
        if self.settings.aws_access_key_id:
            kwargs["aws_access_key_id"] = self.settings.aws_access_key_id
            kwargs["aws_secret_access_key"] = self.settings.aws_secret_access_key
        return kwargs

    def get_client(self, service: str) -> Any:
        if service not in self._clients:
            self._clients[service] = boto3.client(service, **self._base_kwargs)
        return self._clients[service]

    @property
    def s3(self) -> Any:
        return self.get_client("s3")

    @property
    def sagemaker(self) -> Any:
        if not self.settings.feature_sagemaker_enabled:
            raise SageMakerDisabledError(
                f"SageMaker no está habilitado en entorno "
                f"'{self.settings.environment}'. "
                f"Setea FEATURE_SAGEMAKER_ENABLED=true para habilitarlo."
            )
        return self.get_client("sagemaker-runtime")


# Test
from config.settings import Settings, EnvironmentName

local = Settings(environment=EnvironmentName.LOCAL, feature_sagemaker_enabled=False)
factory = ClientFactory(local)

# S3 funciona
print(f"S3 client: {factory.s3}")

# SageMaker lanza error
try:
    factory.sagemaker
except SageMakerDisabledError as e:
    print(f"✅ Esperado: {e}")

# Habilitado en staging
staging = Settings(
    environment=EnvironmentName.STAGING,
    feature_sagemaker_enabled=True,
)
factory_staging = ClientFactory(staging)
print(f"SageMaker client: {factory_staging.sagemaker}")

Ejercicio 2: Service registry pattern

Implementa un ServiceRegistry donde los servicios se registran con un nombre y se resuelven por ese nombre. Útil cuando tienes múltiples implementaciones del mismo servicio.

Ver solución

from typing import Any, Callable


class ServiceNotFoundError(Exception):
    pass


class ServiceRegistry:
    """Registry donde servicios se registran y resuelven por nombre."""

    def __init__(self):
        self._factories: dict[str, Callable] = {}
        self._instances: dict[str, Any] = {}

    def register(self, name: str, factory: Callable):
        """Registra una factory function para un servicio."""
        self._factories[name] = factory

    def resolve(self, name: str) -> Any:
        """Resuelve un servicio por nombre (lazy, singleton)."""
        if name not in self._instances:
            if name not in self._factories:
                raise ServiceNotFoundError(
                    f"Servicio '{name}' no registrado. "
                    f"Disponibles: {list(self._factories.keys())}"
                )
            self._instances[name] = self._factories[name]()
        return self._instances[name]

    def reset(self, name: str | None = None):
        """Resetea instancias cacheadas."""
        if name:
            self._instances.pop(name, None)
        else:
            self._instances.clear()

    @property
    def registered(self) -> list[str]:
        return list(self._factories.keys())


# Uso
from config.settings import Settings, EnvironmentName
from clients.factory import ClientFactory
from services.document_processor import DocumentProcessor

settings = Settings(environment=EnvironmentName.LOCAL)
factory = ClientFactory(settings)

registry = ServiceRegistry()

registry.register(
    "processor",
    lambda: DocumentProcessor(s3_client=factory.s3, bucket=settings.s3_bucket),
)

registry.register(
    "inference",
    lambda: InferenceService(
        lambda_client=factory.lambda_client,
        function_name=settings.lambda_function_name,
    ),
)

print(f"Servicios registrados: {registry.registered}")

processor = registry.resolve("processor")
print(f"Processor: {processor}")

# Resolver de nuevo retorna la misma instancia
processor2 = registry.resolve("processor")
assert processor is processor2
print("✅ Singleton verificado")

Ejercicio 3: Client pool para operaciones paralelas

Crea un ClientPool que mantenga múltiples instancias de un client S3 para permitir operaciones concurrentes (útil para bulk uploads).

Ver solución

import boto3
from typing import Any
from queue import Queue
from contextlib import contextmanager
from config.settings import Settings


class ClientPool:
    """Pool de boto3 clients para operaciones concurrentes."""

    def __init__(self, settings: Settings, service: str, pool_size: int = 5):
        self.settings = settings
        self.service = service
        self._pool: Queue = Queue(maxsize=pool_size)

        kwargs = {"region_name": settings.aws_region}
        if settings.aws_endpoint_url:
            kwargs["endpoint_url"] = settings.aws_endpoint_url
        if settings.aws_access_key_id:
            kwargs["aws_access_key_id"] = settings.aws_access_key_id
            kwargs["aws_secret_access_key"] = settings.aws_secret_access_key

        for _ in range(pool_size):
            self._pool.put(boto3.client(service, **kwargs))

    @contextmanager
    def get_client(self):
        """Obtiene un client del pool (context manager)."""
        client = self._pool.get()
        try:
            yield client
        finally:
            self._pool.put(client)

    @property
    def available(self) -> int:
        return self._pool.qsize()


# Uso
from concurrent.futures import ThreadPoolExecutor
import json

settings = Settings(
    environment="local",
    aws_endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    s3_bucket="pool-test",
)

pool = ClientPool(settings, "s3", pool_size=3)
print(f"Clients disponibles: {pool.available}")


def upload_document(doc_id: int):
    with pool.get_client() as s3:
        s3.put_object(
            Bucket=settings.s3_bucket,
            Key=f"documents/doc-{doc_id:04d}.json",
            Body=json.dumps({"id": doc_id, "data": f"Document {doc_id}"}).encode(),
        )
    return doc_id


# Upload paralelo usando el pool
with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(upload_document, range(10)))

print(f"Documentos subidos: {len(results)}")
print(f"Clients disponibles después: {pool.available}")

Ejercicio 4: Factory con circuit breaker integrado

Extiende ClientFactory para que el health_check() active un circuit breaker si un servicio falla repetidamente, retornando un client "noop" que no hace nada en lugar de un client real.

Ver solución

import time
import boto3
from typing import Any
from config.settings import Settings


class NoopClient:
    """Client que no hace nada — retorna respuestas vacías."""

    def __init__(self, service: str, reason: str):
        self._service = service
        self._reason = reason

    def __getattr__(self, name):
        def noop(*args, **kwargs):
            raise RuntimeError(
                f"Servicio '{self._service}' deshabilitado: {self._reason}"
            )
        return noop


class CircuitBreakerFactory:
    """Factory con circuit breaker por servicio."""

    def __init__(self, settings: Settings, failure_threshold: int = 3, reset_timeout: int = 60):
        self.settings = settings
        self._failure_threshold = failure_threshold
        self._reset_timeout = reset_timeout
        self._failures: dict[str, int] = {}
        self._last_failure: dict[str, float] = {}
        self._clients: dict[str, Any] = {}

        self._base_kwargs = {"region_name": settings.aws_region}
        if settings.aws_endpoint_url:
            self._base_kwargs["endpoint_url"] = settings.aws_endpoint_url
        if settings.aws_access_key_id:
            self._base_kwargs["aws_access_key_id"] = settings.aws_access_key_id
            self._base_kwargs["aws_secret_access_key"] = settings.aws_secret_access_key

    def get_client(self, service: str) -> Any:
        failures = self._failures.get(service, 0)
        last_fail = self._last_failure.get(service, 0)

        if failures >= self._failure_threshold:
            elapsed = time.time() - last_fail
            if elapsed < self._reset_timeout:
                return NoopClient(
                    service,
                    f"Circuit open: {failures} failures, "
                    f"reset in {self._reset_timeout - elapsed:.0f}s",
                )
            self._failures[service] = 0

        if service not in self._clients:
            self._clients[service] = boto3.client(service, **self._base_kwargs)
        return self._clients[service]

    def record_failure(self, service: str):
        self._failures[service] = self._failures.get(service, 0) + 1
        self._last_failure[service] = time.time()

    def record_success(self, service: str):
        self._failures[service] = 0

    def circuit_status(self) -> dict:
        status = {}
        for service in set(list(self._clients.keys()) + list(self._failures.keys())):
            failures = self._failures.get(service, 0)
            if failures >= self._failure_threshold:
                elapsed = time.time() - self._last_failure.get(service, 0)
                status[service] = {
                    "state": "open",
                    "failures": failures,
                    "reset_in": max(0, self._reset_timeout - elapsed),
                }
            elif failures > 0:
                status[service] = {"state": "half-open", "failures": failures}
            else:
                status[service] = {"state": "closed", "failures": 0}
        return status


# Demo
settings = Settings(environment="local")
factory = CircuitBreakerFactory(settings, failure_threshold=2, reset_timeout=10)

s3 = factory.get_client("s3")
print(f"S3 client: {type(s3).__name__}")

factory.record_failure("s3")
factory.record_failure("s3")

s3_after = factory.get_client("s3")
print(f"S3 after 2 failures: {type(s3_after).__name__}")

print(f"Circuit status: {factory.circuit_status()}")

Resumen

Dependency injection desacopla tu lógica de negocio de la infraestructura. DocumentProcessor no sabe si habla con LocalStack, AWS, o un mock. Recibe un client y trabaja.
El factory pattern centraliza la construcción de clients. Un solo lugar sabe cómo crear boto3 clients para el entorno actual. El resto de la app solo pide clients.
DI hace tus tests rápidos y confiables. Con mock clients, testeas lógica de negocio en millisegundos sin infraestructura.
El ServiceContainer organiza el wiring. Es donde config + factory + servicios se conectan. El handler de Lambda solo necesita el container.
Clients resilientes agregan retry sin cambiar lógica de negocio. El wrapper es transparente — tu servicio no sabe que hay retry.
En la siguiente cápsula, usarás estos patrones para testing multi-entorno — mismo test suite contra LocalStack y AWS.

Recursos Adicionales

Martin Fowler — Dependency Injection — El artículo fundacional sobre DI
Factory Method Pattern — Referencia del patrón factory
python-dependency-injector — Framework DI para Python
boto3 Client Reference — Cómo se crean clients en boto3
unittest.mock — Python Docs — Mocking en Python
Circuit Breaker Pattern — Patrón para resilience

← Anterior3. Config Management Multi-Entorno Siguiente →5. Testing Multi-Entorno