Hexagonal Architecture
DDD, puertos y adaptadores, capas domain/application/infrastructure y SOLID.
La arquitectura hexagonal (también llamada Ports & Adapters, propuesta por Alistair Cockburn) parte de una premisa simple pero poderosa: el núcleo de tu sistema —la lógica de negocio— no debería saber nada del mundo exterior. No sabe si los datos vienen de una base de datos PostgreSQL, de un CSV en S3, o de un JSON via HTTP. Esta ignorancia deliberada es la que hace al sistema testeable, mantenible y adaptable a cambios tecnológicos.
En el contexto de ML en producción, la arquitectura hexagonal separa tres capas. Domain: las entidades y reglas de negocio puras (ChurnScore, CustomerFeatures, ThresholdPolicy). Application: los casos de uso que orquestan la lógica (PredictChurnUseCase). Infrastructure: los adaptadores concretos que conectan con el mundo exterior (FastAPIAdapter, SklearnModelAdapter, PostgresRepository). La regla de dependencia es unidireccional: infrastructure depende de application, application depende de domain, domain no depende de nadie.
Los puertos son interfaces (ABC o Protocol en Python) que el dominio define como contratos. El puerto ModelPort define el método predict(features) sin saber si por detrás hay sklearn, XGBoost, o una llamada a SageMaker. Los adaptadores son las implementaciones concretas de esos puertos. El domain no importa jamás desde infrastructure: esa dirección de dependencia está prohibida por diseño.
SOLID se aplica naturalmente en esta arquitectura. Single Responsibility: cada clase tiene una sola razón para cambiar. Open/Closed: cuando el modelo de ML cambia de sklearn a XGBoost, creas un nuevo adaptador sin tocar el caso de uso. Dependency Inversion: el caso de uso depende del puerto abstracto (ModelPort), no de la implementación concreta (SklearnModelAdapter). Esto permite inyectar mocks en los tests sin tocar el código de producción.
Un ejemplo de productivización real con este patrón: el proyecto de churn en https://churn.benjacode.com usa FastAPI como adaptador de entrada, SageMaker como adaptador de modelo, y Lambda como infraestructura de cómputo. El caso de uso de predicción no sabe nada de ninguno de estos tres: puede testearse con mocks en milisegundos, sin llamadas reales a AWS ni a la API.
El beneficio más tangible es la velocidad de los tests. Con arquitectura hexagonal, el 90% de la lógica de negocio se testea con unit tests puros: sin base de datos, sin red, sin modelo real. Los tests de integración que prueban los adaptadores reales son pocos y corren en el pipeline de CI. Esto cambia el ciclo de feedback de minutos a segundos.
# Project structure
src/
domain/
entities.py # ChurnScore, CustomerFeatures
ports.py # ModelPort, FeatureRepository (ABCs)
application/
use_cases.py # PredictChurnUseCase
infrastructure/
adapters/
sklearn_model.py # SklearnModelAdapter implements ModelPort
fastapi_api.py # FastAPI routes
postgres_repo.py # PostgresFeatureRepository
# domain/ports.py — pure interfaces, no imports from infra
from abc import ABC, abstractmethod
from dataclasses import dataclass
@dataclass
class CustomerFeatures:
age: int
tenure: float
monthly_charges: float
@dataclass
class ChurnScore:
probability: float
label: str
class ModelPort(ABC):
@abstractmethod
def predict(self, features: CustomerFeatures) -> ChurnScore: ...
# application/use_cases.py
class PredictChurnUseCase:
def __init__(self, model: ModelPort) -> None:
self._model = model
def execute(self, features: CustomerFeatures) -> ChurnScore:
return self._model.predict(features)
# infrastructure/adapters/sklearn_model.py
import joblib
from src.domain.ports import CustomerFeatures, ChurnScore, ModelPort
class SklearnModelAdapter(ModelPort):
def __init__(self, model_path: str) -> None:
self._model = joblib.load(model_path)
def predict(self, features: CustomerFeatures) -> ChurnScore:
X = [[features.age, features.tenure, features.monthly_charges]]
prob = self._model.predict_proba(X)[0][1]
return ChurnScore(
probability=round(prob, 4),
label='churn' if prob > 0.5 else 'no_churn',
)Debugging lab
Detecta y corrige el error en el código.
- 5.3.5.1
# domain/ports.py import joblib from pathlib import Path class ModelPort: def __init__(self): self._model = joblib.load(Path('model.pkl')) def predict(self, features: list[float]) -> float: return self._model.predict_proba([features])[0][1]
- 5.3.5.2
# application/use_cases.py import joblib from src.domain.ports import CustomerFeatures, ChurnScore class PredictChurnUseCase: def __init__(self, model_path: str) -> None: self._model = joblib.load(model_path) def execute(self, features: CustomerFeatures) -> ChurnScore: prob = self._model.predict_proba([[features.age, features.tenure]])[0][1] return ChurnScore(probability=prob, label='churn' if prob > 0.5 else 'no_churn')
- 5.3.5.3
# infrastructure/adapters/fastapi_api.py import joblib from fastapi import FastAPI from src.domain.ports import CustomerFeatures app = FastAPI() model = joblib.load('model.pkl') @app.post('/predict') def predict(age: int, tenure: float): features = [[age, tenure]] prob = model.predict_proba(features)[0][1] label = 'churn' if prob > 0.5 else 'no_churn' return {'probability': prob, 'label': label}
- 5.3.5.4
# Test del caso de uso import joblib from src.application.use_cases import PredictChurnUseCase from src.domain.ports import CustomerFeatures def test_predict_churn(): use_case = PredictChurnUseCase(model_path='tests/fixtures/model.pkl') features = CustomerFeatures(age=35, tenure=12.0, monthly_charges=75.0) score = use_case.execute(features) assert score.probability > 0
- 5.3.5.5
# domain/entities.py from pydantic import BaseModel from sqlalchemy import Column, Integer, Float from sqlalchemy.orm import declarative_base Base = declarative_base() class CustomerFeatures(Base, BaseModel): __tablename__ = 'customers' age: int = Column(Integer) tenure: float = Column(Float)