Patrones de Diseño Python

ANTES lead_processor.py (780 lineas) todo mezclado

# Un metodo con 5 responsabilidades mezcladas

class LeadProcessor:
    async def process_lead(
        self, request: Request
    ) -> Response:
        # 1. Parse HTTP (deberia ser del handler)
        data = await request.json()

        # 2. Validacion inline (deberia ser Pydantic)
        if not data.get("email"):
            return Response({"error": "email required"}, 400)
        if not data.get("company"):
            return Response({"error": "company required"}, 400)

        # 3. Scoring hardcoded (deberia ser ScoringEngine)
        score = 0
        if data.get("company_size", 0) > 50: score += 30
        if data.get("budget", 0) > 5000: score += 40
        if data.get("timeline") == "immediate": score += 30

        # 4. SQL directo (deberia ser Repository)
        lead = await db.execute(
            "INSERT INTO leads (email, company, score)"
            " VALUES ($1, $2, $3) RETURNING *",
            data["email"], data["company"], score
        )

        # 5. Notificaciones acopladas (imposible testear)
        smtp = SmtpClient(host="smtp.sendgrid.com")
        smtp.send(data["email"], f"score: {score}")
        slack_webhook.post(f"Nuevo lead: score {score}")

        return Response({"lead_id": lead.id}, 201)

DESPUES lead_handler.py (50 lineas) solo HTTP

# Handler: solo HTTP. Sin logica de negocio.

class LeadHandler:
    """HTTP layer: parse, validate shape, format response."""

    def __init__(self, service: LeadService) -> None:
        # INYECCION DE DEPENDENCIAS — testeable
        self._service = service

    @app.post("/leads", status_code=201)
    async def create_lead(
        self,
        payload: LeadRequest,  # Pydantic valida
    ) -> LeadResponse:
        # Una sola responsabilidad: orquestar HTTP
        lead = await self._service.create_lead(
            CreateLeadInput.from_request(payload)
        )
        return LeadResponse.from_lead(lead)

# ------------------------------------------
# lead_service.py — solo logica de negocio

class ScoringEngine:
    """Calcula score. Pura, testeable, sin I/O."""

    def score(self, data: CreateLeadInput) -> int:
        points = 0
        if data.company_size > 50: points += 30
        if data.budget > 5000: points += 40
        if data.timeline == "immediate": points += 30
        return points

class LeadService:
    def __init__(
        self,
        repo: LeadRepository,       # inyectado
        scorer: ScoringEngine,      # inyectado
        notifier: NotificationService, # inyectado
    ) -> None:
        self._repo = repo
        self._scorer = scorer
        self._notifier = notifier

    async def create_lead(
        self, data: CreateLeadInput
    ) -> Lead:
        score = self._scorer.score(data)
        lead = await self._repo.save(
            Lead(email=data.email, score=score)
        )
        await self._notifier.notify(lead)
        return lead

PATRON COMPOSICION notifications.py Protocols + constructor injection

from typing import Protocol

# Protocols: define contratos minimos, no herencia
class EmailSender(Protocol):
    async def send(self, to: str, message: str) -> None: ...

class SlackSender(Protocol):
    async def post(self, channel: str, message: str) -> None: ...


class NotificationService:
    """Composicion de canales. KISS: sin factory, sin registry."""

    def __init__(
        self,
        email: EmailSender,
        slack: SlackSender | None = None,  # opcional
    ) -> None:
        self._email = email
        self._slack = slack

    async def notify(self, lead: Lead) -> None:
        msg = f"Nuevo lead {lead.email} — score {lead.score}/100"
        await self._email.send(lead.email, msg)
        if self._slack:
            await self._slack.post("#leads", msg)


# PRODUCCION: senders reales
notification_service = NotificationService(
    email=SendGridEmailSender(api_key=settings.SENDGRID_KEY),
    slack=SlackWebhookSender(url=settings.SLACK_WEBHOOK_URL),
)

# TESTS: fakes que implementan el Protocol
class FakeEmailSender:
    sent: list[str] = []
    async def send(self, to, msg): self.sent.append(to)

notification_service_test = NotificationService(
    email=FakeEmailSender(),   # sin SMTP real
    slack=None,                # Slack desactivado en test
)

KISSSin factory registry para formatters

El monolito tenia la logica para "elegir canal" con un switch gigante. Lo reemplazamos por un simple dict de senders. La complejidad no estaba justificada porque los canales no cambian en runtime. Regla: si no tienes 3+ casos de extension real, usa lo mas simple que funcione.

SRPScoringEngine separado del Service

El scoring cambio 4 veces en 6 meses (reglas de negocio). El guardado en BD cambio 1 vez (infraestructura). Al tener una sola razon de cambio cada uno, podemos testear ScoringEngine con datos puros sin tocar la BD, y cambiar reglas sin riesgo de romper el guardado.

INY. DEP.Constructor injection en todas las clases

LeadService recibe repo, scorer y notifier por constructor. En tests: InMemoryRepo, FakeScoringEngine, FakeNotifier. Cero hits a PostgreSQL ni a SendGrid. Los tests corren en 50ms en lugar de 3 segundos. El constructor con 3 params sigue siendo manejable; si crece a 6+ es senal de nueva clase.

COMPComposicion para notificaciones

Antes: EmailNotificationService(SmtpClient()) — herencia rigida imposible de mockear. Ahora: NotificationService(email=SendGridSender(), slack=SlackSender()) — se pasan como parametros. Anadir WhatsApp o Discord en el futuro es agregar un nuevo sender que implementa el Protocol, sin tocar NotificationService.

Arquitectura de Capas

KISSSin factory registry para formatters

SRPScoringEngine separado del Service

INY. DEP.Constructor injection en todas las clases

COMPComposicion para notificaciones