Patrones de diseño de Python para una arquitectura limpia
Construya aplicaciones Python mantenibles con patrones de diseño SOLID
Arquitectura Limpia ha revolucionado la forma en que los desarrolladores construyen aplicaciones escalables y mantenibles, enfatizando la separación de preocupaciones y la gestión de dependencias.
En Python, estos principios se combinan con la naturaleza dinámica del lenguaje para crear sistemas flexibles y probables que evolucionan con los requisitos del negocio sin convertirse en deuda técnica.

Entendiendo la Arquitectura Limpia en Python
La Arquitectura Limpia, introducida por Robert C. Martin (Uncle Bob), organiza el software en capas concéntricas donde las dependencias apuntan hacia adentro hacia la lógica empresarial central. Este patrón arquitectónico asegura que las reglas empresariales críticas de su aplicación permanezcan independientes de los marcos, bases de datos y servicios externos.
La Filosofía Central
El principio fundamental es simple pero poderoso: la lógica empresarial no debe depender de la infraestructura. Sus entidades empresariales, casos de uso y reglas empresariales deben funcionar independientemente de si está utilizando PostgreSQL o MongoDB, FastAPI o Flask, AWS o Azure.
En Python, esta filosofía se alinea perfectamente con la “tipificación de pato” y la programación orientada a protocolos del lenguaje, permitiendo una separación limpia sin la ceremonia requerida en lenguajes tipificados estáticamente.
Las Cuatro Capas de la Arquitectura Limpia
Capa de Entidades (Dominio): Objetos empresariales puros con reglas empresariales de alcance empresarial. Estos son POPOs (Plain Old Python Objects) sin dependencias externas.
Capa de Casos de Uso (Aplicación): Reglas empresariales específicas de la aplicación que orquestan el flujo de datos entre entidades y servicios externos.
Capa de Adaptadores de Interfaz: Convierte datos entre el formato más conveniente para los casos de uso y entidades, y el formato requerido por agencias externas.
Capa de Marcos y Conducentes: Todos los detalles externos como bases de datos, marcos web y APIs externas.
Principios SOLID en Python
Los principios SOLID forman la base de la arquitectura limpia. Vamos a explorar cómo cada principio se manifiesta en Python. Para una visión general completa de los patrones de diseño en Python, consulte la Guía de Patrones de Diseño en Python.
Principio de Responsabilidad Única (SRP)
Cada clase debe tener un único motivo para cambiar:
# Mal: Varias responsabilidades
class UserManager:
def create_user(self, user_data):
# Crear usuario
pass
def send_welcome_email(self, user):
# Enviar correo
pass
def log_creation(self, user):
# Registrar en archivo
pass
# Bien: Responsabilidades separadas
class UserService:
def __init__(self, repository, email_service, logger):
self.repository = repository
self.email_service = email_service
self.logger = logger
def create_user(self, user_data):
user = User(**user_data)
self.repository.save(user)
self.email_service.send_welcome(user)
self.logger.info(f"Usuario creado: {user.id}")
return user
Principio de Abierto/Cerrado (OCP)
Las entidades de software deben estar abiertas para su extensión pero cerradas para su modificación:
from abc import ABC, abstractmethod
from typing import Protocol
# Usando Protocol (Python 3.8+)
class PaymentProcessor(Protocol):
def process_payment(self, amount: float) -> bool:
...
class CreditCardProcessor:
def process_payment(self, amount: float) -> bool:
# Lógica de tarjeta de crédito
return True
class PayPalProcessor:
def process_payment(self, amount: float) -> bool:
# Lógica de PayPal
return True
# Fácilmente extensible sin modificar el código existente
class CryptoProcessor:
def process_payment(self, amount: float) -> bool:
# Lógica de criptomonedas
return True
Principio de Sustitución de Liskov (LSP)
Los objetos deben ser sustituibles por sus subtipos sin romper el programa:
from abc import ABC, abstractmethod
class DataStore(ABC):
@abstractmethod
def save(self, key: str, value: str) -> None:
pass
@abstractmethod
def get(self, key: str) -> str:
pass
class PostgreSQLStore(DataStore):
def save(self, key: str, value: str) -> None:
# Implementación de PostgreSQL
pass
def get(self, key: str) -> str:
# Implementación de PostgreSQL
return ""
class RedisStore(DataStore):
def save(self, key: str, value: str) -> None:
# Implementación de Redis
pass
def get(self, key: str) -> str:
# Implementación de Redis
return ""
# Ambos pueden usarse intercambiablemente
def process_data(store: DataStore, key: str, value: str):
store.save(key, value)
return store.get(key)
Principio de Segregación de Interfaces (ISP)
Los clientes no deben verse obligados a depender de interfaces que no utilizan:
# Mal: Interfaz muy gruesa
class Worker(ABC):
@abstractmethod
def work(self): pass
@abstractmethod
def eat(self): pass
@abstractmethod
def sleep(self): pass
# Bien: Interfaces segregadas
class Workable(Protocol):
def work(self) -> None: ...
class Eatable(Protocol):
def eat(self) -> None: ...
class Human:
def work(self) -> None:
print("Trabajando")
def eat(self) -> None:
print("Comiendo")
class Robot:
def work(self) -> None:
print("Trabajando")
# No se necesita el método eat
Principio de Inversión de Dependencias (DIP)
Los módulos de alto nivel no deben depender de módulos de bajo nivel. Ambos deben depender de abstracciones:
from typing import Protocol
# Abstracción
class EmailSender(Protocol):
def send(self, to: str, subject: str, body: str) -> None:
...
# Módulo de bajo nivel
class SMTPEmailSender:
def send(self, to: str, subject: str, body: str) -> None:
# Implementación SMTP
pass
# Módulo de alto nivel depende de abstracción
class UserRegistrationService:
def __init__(self, email_sender: EmailSender):
self.email_sender = email_sender
def register(self, email: str, name: str):
# Lógica de registro
self.email_sender.send(
to=email,
subject="¡Bienvenido!",
body=f"Hola {name}"
)
Patrón de Repositorio: Abstrayendo el Acceso a Datos
El Patrón de Repositorio proporciona una interfaz similar a una colección para acceder a objetos del dominio, ocultando los detalles del almacenamiento de datos.
Implementación Básica de Repositorio
from abc import ABC, abstractmethod
from typing import List, Optional
from dataclasses import dataclass
from uuid import UUID, uuid4
@dataclass
class User:
id: UUID
email: str
name: str
is_active: bool = True
class UserRepository(ABC):
@abstractmethod
def save(self, user: User) -> User:
pass
@abstractmethod
def get_by_id(self, user_id: UUID) -> Optional[User]:
pass
@abstractmethod
def get_by_email(self, email: str) -> Optional[User]:
pass
@abstractmethod
def list_all(self) -> List[User]:
pass
@abstractmethod
def delete(self, user_id: UUID) -> bool:
pass
Implementación con SQLAlchemy
from sqlalchemy import create_engine, Column, String, Boolean
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
Base = declarative_base()
class UserModel(Base):
__tablename__ = 'users'
id = Column(PGUUID(as_uuid=True), primary_key=True)
email = Column(String, unique=True, nullable=False)
name = Column(String, nullable=False)
is_active = Column(Boolean, default=True)
class SQLAlchemyUserRepository(UserRepository):
def __init__(self, session: Session):
self.session = session
def save(self, user: User) -> User:
user_model = UserModel(
id=user.id,
email=user.email,
name=user.name,
is_active=user.is_active
)
self.session.add(user_model)
self.session.commit()
return user
def get_by_id(self, user_id: UUID) -> Optional[User]:
user_model = self.session.query(UserModel).filter(
UserModel.id == user_id
).first()
if not user_model:
return None
return User(
id=user_model.id,
email=user_model.email,
name=user_model.name,
is_active=user_model.is_active
)
def get_by_email(self, email: str) -> Optional[User]:
user_model = self.session.query(UserModel).filter(
UserModel.email == email
).first()
if not user_model:
return None
return User(
id=user_model.id,
email=user_model.email,
name=user_model.name,
is_active=user_model.is_active
)
def list_all(self) -> List[User]:
users = self.session.query(UserModel).all()
return [
User(
id=u.id,
email=u.email,
name=u.name,
is_active=u.is_active
)
for u in users
]
def delete(self, user_id: UUID) -> bool:
result = self.session.query(UserModel).filter(
UserModel.id == user_id
).delete()
self.session.commit()
return result > 0
Repositorio en Memoria para Pruebas
class InMemoryUserRepository(UserRepository):
def __init__(self):
self.users: dict[UUID, User] = {}
def save(self, user: User) -> User:
self.users[user.id] = user
return user
def get_by_id(self, user_id: UUID) -> Optional[User]:
return self.users.get(user_id)
def get_by_email(self, email: str) -> Optional[User]:
for user in self.users.values():
if user.email == email:
return user
return None
def list_all(self) -> List[User]:
return list(self.users.values())
def delete(self, user_id: UUID) -> bool:
if user_id in self.users:
del self.users[user_id]
return True
return False
Capa de Servicio: Orquestando la Lógica Empresarial
La Capa de Servicio implementa casos de uso y orquesta el flujo entre repositorios, servicios externos y lógica empresarial.
from typing import Optional
from uuid import uuid4
class UserAlreadyExistsError(Exception):
pass
class UserNotFoundError(Exception):
pass
class UserService:
def __init__(
self,
user_repository: UserRepository,
email_service: EmailSender,
event_publisher: 'EventPublisher'
):
self.user_repository = user_repository
self.email_service = email_service
self.event_publisher = event_publisher
def register_user(self, email: str, name: str) -> User:
# Verificar si el usuario existe
existing_user = self.user_repository.get_by_email(email)
if existing_user:
raise UserAlreadyExistsError(f"El usuario con el correo {email} ya existe")
# Crear nuevo usuario
user = User(
id=uuid4(),
email=email,
name=name,
is_active=True
)
# Guardar en el repositorio
user = self.user_repository.save(user)
# Enviar correo de bienvenida
self.email_service.send(
to=user.email,
subject="¡Bienvenido!",
body=f"Hola {user.name}, ¡bienvenido a nuestra plataforma!"
)
# Publicar evento
self.event_publisher.publish('user.registered', {
'user_id': str(user.id),
'email': user.email
})
return user
def deactivate_user(self, user_id: UUID) -> User:
user = self.user_repository.get_by_id(user_id)
if not user:
raise UserNotFoundError(f"Usuario {user_id} no encontrado")
user.is_active = False
user = self.user_repository.save(user)
self.event_publisher.publish('user.deactivated', {
'user_id': str(user.id)
})
return user
Inyección de Dependencias en Python
La naturaleza dinámica de Python hace que la inyección de dependencias sea sencilla sin requerir marcos pesados.
Inyección por Constructor
class OrderService:
def __init__(
self,
order_repository: 'OrderRepository',
payment_processor: PaymentProcessor,
notification_service: 'NotificationService'
):
self.order_repository = order_repository
self.payment_processor = payment_processor
self.notification_service = notification_service
def place_order(self, order_data: dict):
# Usar dependencias inyectadas
pass
Contenedor de Dependencias Simple
from typing import Dict, Type, Callable, Any
class Container:
def __init__(self):
self._services: Dict[Type, Callable] = {}
self._singletons: Dict[Type, Any] = {}
def register(self, interface: Type, factory: Callable):
self._services[interface] = factory
def register_singleton(self, interface: Type, instance: Any):
self._singletons[interface] = instance
def resolve(self, interface: Type):
if interface in self._singletons:
return self._singletons[interface]
factory = self._services.get(interface)
if factory:
return factory(self)
raise ValueError(f"No se encontró registro para {interface}")
# Uso
def create_container() -> Container:
container = Container()
# Registrar servicios
container.register_singleton(
Session,
sessionmaker(bind=create_engine('postgresql://...'))()
)
container.register(
UserRepository,
lambda c: SQLAlchemyUserRepository(c.resolve(Session))
)
container.register(
EmailSender,
lambda c: SMTPEmailSender()
)
container.register(
UserService,
lambda c: UserService(
c.resolve(UserRepository),
c.resolve(EmailSender),
c.resolve(EventPublisher)
)
)
return container
Arquitectura Hexagonal (Puertos y Adaptadores)
La Arquitectura Hexagonal coloca la lógica empresarial en el centro con adaptadores que manejan la comunicación externa.
Definiendo Puertos (Interfaces)
# Puerto de Entrada (Primario)
class CreateUserUseCase(Protocol):
def execute(self, request: 'CreateUserRequest') -> 'CreateUserResponse':
...
# Puerto de Salida (Secundario)
class UserPersistencePort(Protocol):
def save(self, user: User) -> User:
...
def find_by_email(self, email: str) -> Optional[User]:
...
Implementando Adaptadores
from pydantic import BaseModel, EmailStr
# Adaptador de Entrada (API REST)
from fastapi import FastAPI, Depends, HTTPException
class CreateUserRequest(BaseModel):
email: EmailStr
name: str
class CreateUserResponse(BaseModel):
id: str
email: str
name: str
app = FastAPI()
@app.post("/users", response_model=CreateUserResponse)
def create_user(
request: CreateUserRequest,
user_service: UserService = Depends(get_user_service)
):
try:
user = user_service.register_user(
email=request.email,
name=request.name
)
return CreateUserResponse(
id=str(user.id),
email=user.email,
name=user.name
)
except UserAlreadyExistsError as e:
raise HTTPException(status_code=400, detail=str(e))
# Adaptador de Salida (Base de datos)
# Ya implementado como SQLAlchemyUserRepository
Patrones de Diseño Orientado al Dominio
Objetos de Valor
Objetos inmutables definidos por sus atributos:
from dataclasses import dataclass
from typing import Pattern
import re
@dataclass(frozen=True)
class Email:
value: str
EMAIL_PATTERN: Pattern = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')
def __post_init__(self):
if not self.EMAIL_PATTERN.match(self.value):
raise ValueError(f"Correo electrónico inválido: {self.value}")
def __str__(self):
return self.value
@dataclass(frozen=True)
class Money:
amount: float
currency: str
def __post_init__(self):
if self.amount < 0:
raise ValueError("El monto no puede ser negativo")
if self.currency not in ['USD', 'EUR', 'GBP']:
raise ValueError(f"Moneda no soportada: {self.currency}")
def add(self, other: 'Money') -> 'Money':
if self.currency != other.currency:
raise ValueError("No se pueden sumar monedas diferentes")
return Money(self.amount + other.amount, self.currency)
Agregados
Clúster de objetos del dominio tratados como una unidad única:
from dataclasses import dataclass, field
from typing import List
from datetime import datetime
@dataclass
class OrderItem:
product_id: UUID
quantity: int
price: Money
def total(self) -> Money:
return Money(
self.price.amount * self.quantity,
self.price.currency
)
@dataclass
class Order:
id: UUID
customer_id: UUID
items: List[OrderItem] = field(default_factory=list)
status: str = "pendiente"
created_at: datetime = field(default_factory=datetime.now)
def add_item(self, product_id: UUID, quantity: int, price: Money):
item = OrderItem(product_id, quantity, price)
self.items.append(item)
def remove_item(self, product_id: UUID):
self.items = [
item for item in self.items
if item.product_id != product_id
]
def total(self) -> Money:
if not self.items:
return Money(0, "USD")
return sum(
(item.total() for item in self.items),
Money(0, self.items[0].price.currency)
)
def confirm(self):
if not self.items:
raise ValueError("No se puede confirmar un pedido vacío")
if self.status != "pendiente":
raise ValueError("El pedido ya fue procesado")
self.status = "confirmado"
Eventos del Dominio
Los eventos del dominio permiten una acoplamiento suelto entre componentes y soportan arquitecturas orientadas a eventos. Para sistemas a gran escala orientados a eventos, considere implementar el streaming de eventos con servicios como AWS Kinesis—consulte Construyendo Microservicios Orientados a Eventos con AWS Kinesis para una guía detallada.
from dataclasses import dataclass
from datetime import datetime
from typing import List, Callable
@dataclass
class DomainEvent:
occurred_at: datetime = field(default_factory=datetime.now)
@dataclass
class OrderConfirmed(DomainEvent):
order_id: UUID
customer_id: UUID
total: Money
class EventPublisher:
def __init__(self):
self._handlers: Dict[Type, List[Callable]] = {}
def subscribe(self, event_type: Type, handler: Callable):
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
def publish(self, event: DomainEvent):
event_type = type(event)
handlers = self._handlers.get(event_type, [])
for handler in handlers:
handler(event)
Características modernas de Python para una arquitectura limpia
Las características modernas de Python hacen que la implementación de una arquitectura limpia sea más elegante y segura en cuanto a tipos. Si necesitas una referencia rápida sobre la sintaxis y características de Python, consulta el Python Cheatsheet.
Tipos y protocolos
from typing import Protocol, runtime_checkable
@runtime_checkable
class Serializable(Protocol):
def to_dict(self) -> dict:
...
@classmethod
def from_dict(cls, data: dict) -> 'Serializable':
...
def serialize(obj: Serializable) -> dict:
return obj.to_dict()
Pydantic para validación
from pydantic import BaseModel, Field, validator
from typing import Optional
class CreateUserDTO(BaseModel):
email: EmailStr
name: str = Field(..., min_length=2, max_length=100)
age: Optional[int] = Field(None, ge=0, le=150)
@validator('name')
def name_must_not_contain_numbers(cls, v):
if any(char.isdigit() for char in v):
raise ValueError('El nombre no puede contener números')
return v
class Config:
frozen = True # Hacerlo inmutable
Async/Await para operaciones de E/S
La sintaxis async/await de Python es especialmente poderosa para operaciones de E/S en una arquitectura limpia, permitiendo interacciones no bloqueantes con bases de datos y servicios externos. Al desplegar aplicaciones de Python en plataformas sin servidor, entender las características de rendimiento se vuelve crucial—consulta AWS lambda performance: JavaScript vs Python vs Golang para obtener información sobre la optimización de funciones sin servidor en Python.
from typing import List
import asyncio
class AsyncUserRepository(ABC):
@abstractmethod
async def save(self, user: User) -> User:
pass
@abstractmethod
async def get_by_id(self, user_id: UUID) -> Optional[User]:
pass
class AsyncUserService:
def __init__(self, repository: AsyncUserRepository):
self.repository = repository
async def register_user(self, email: str, name: str) -> User:
user = User(id=uuid4(), email=email, name=name)
return await self.repository.save(user)
async def get_users_batch(self, user_ids: List[UUID]) -> List[User]:
tasks = [self.repository.get_by_id(uid) for uid in user_ids]
results = await asyncio.gather(*tasks)
return [u for u in results if u is not None]
Mejores prácticas para la estructura del proyecto
Una organización adecuada del proyecto es esencial para mantener una arquitectura limpia. Antes de configurar la estructura de tu proyecto, asegúrate de usar entornos virtuales de Python para aislar las dependencias. El venv Cheatsheet cubre todo lo que necesitas saber sobre la gestión de entornos virtuales. Para proyectos modernos de Python, considera usar uv - Nuevo gestor de paquetes, proyectos y entornos de Python, que proporciona una gestión más rápida de paquetes y configuración de proyectos.
my_application/
├── domain/ # Reglas empresariales del núcleo
│ ├── __init__.py
│ ├── entities/
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── order.py
│ ├── value_objects/
│ │ ├── __init__.py
│ │ ├── email.py
│ │ └── money.py
│ ├── events/
│ │ ├── __init__.py
│ │ └── user_events.py
│ └── exceptions.py
│
├── application/ # Reglas empresariales de la aplicación
│ ├── __init__.py
│ ├── use_cases/
│ │ ├── __init__.py
│ │ ├── create_user.py
│ │ └── place_order.py
│ ├── services/
│ │ ├── __init__.py
│ │ └── user_service.py
│ └── ports/
│ ├── __init__.py
│ ├── repositories.py
│ └── external_services.py
│
├── infrastructure/ # Interfaces externas
│ ├── __init__.py
│ ├── persistence/
│ │ ├── __init__.py
│ │ ├── sqlalchemy/
│ │ │ ├── models.py
│ │ │ └── repositories.py
│ │ └── mongodb/
│ │ └── repositories.py
│ ├── messaging/
│ │ ├── __init__.py
│ │ └── rabbitmq_publisher.py
│ ├── external_services/
│ │ ├── __init__.py
│ │ └── email_service.py
│ └── config.py
│
├── presentation/ # Capa de UI/API
│ ├── __init__.py
│ ├── api/
│ │ ├── __init__.py
│ │ ├── dependencies.py
│ │ ├── routes/
│ │ │ ├── __init__.py
│ │ │ ├── users.py
│ │ │ └── orders.py
│ │ └── schemas/
│ │ ├── __init__.py
│ │ └── user_schemas.py
│ └── cli/
│ └── commands.py
│
├── tests/
│ ├── unit/
│ ├── integration/
│ └── e2e/
│
├── main.py # Punto de entrada de la aplicación
├── container.py # Configuración de inyección de dependencias
├── pyproject.toml
└── README.md
Pruebas de arquitectura limpia
Pruebas unitarias de lógica del dominio
import pytest
from uuid import uuid4
def test_user_creation():
user = User(
id=uuid4(),
email="test@example.com",
name="Test User"
)
assert user.email == "test@example.com"
assert user.is_active is True
def test_order_total_calculation():
order = Order(id=uuid4(), customer_id=uuid4())
order.add_item(
uuid4(),
quantity=2,
price=Money(10.0, "USD")
)
order.add_item(
uuid4(),
quantity=1,
price=Money(5.0, "USD")
)
assert order.total().amount == 25.0
Pruebas de integración con repositorio
@pytest.fixture
def in_memory_repository():
return InMemoryUserRepository()
def test_user_repository_save_and_retrieve(in_memory_repository):
user = User(
id=uuid4(),
email="test@example.com",
name="Test User"
)
saved_user = in_memory_repository.save(user)
retrieved_user = in_memory_repository.get_by_id(user.id)
assert retrieved_user is not None
assert retrieved_user.email == user.email
Pruebas de la capa de servicios
from unittest.mock import Mock
def test_user_registration():
# Arrange
mock_repository = Mock(spec=UserRepository)
mock_repository.get_by_email.return_value = None
mock_repository.save.return_value = User(
id=uuid4(),
email="test@example.com",
name="Test"
)
mock_email = Mock(spec=EmailSender)
mock_events = Mock(spec=EventPublisher)
service = UserService(mock_repository, mock_email, mock_events)
# Act
user = service.register_user("test@example.com", "Test")
# Assert
assert user.email == "test@example.com"
mock_repository.save.assert_called_once()
mock_email.send.assert_called_once()
mock_events.publish.assert_called_once()
Errores comunes y cómo evitarlos
Sobrediseño
No implementes una arquitectura limpia para aplicaciones CRUD simples. Comienza sencillo y refactoriza a medida que crece la complejidad.
Abstracciones permeables
Asegúrate de que las entidades del dominio no contengan anotaciones de base de datos o código específico del framework:
# Mal
from sqlalchemy import Column
@dataclass
class User:
id: Column(Integer, primary_key=True) # Código del framework filtrando al dominio
# Bien
@dataclass
class User:
id: UUID # Objeto puro del dominio
Dependencias circulares
Usa inyección de dependencias e interfaces para romper dependencias circulares entre capas.
Ignorar el contexto
La arquitectura limpia no es un todo o nada. Ajusta la rigidez de las capas según el tamaño del proyecto y la experiencia del equipo.
Enlaces útiles
- Arquitectura limpia por Robert C. Martin
- Documentación de Tipos de Python
- Documentación de Pydantic
- Documentación oficial de FastAPI
- Documentación de ORM de SQLAlchemy
- Biblioteca de Inyección de Dependencias
- Referencia de Diseño Orientado a Dominios
- Patrones de Arquitectura con Python
- Blog de Martin Fowler sobre Arquitectura
- Guía de patrones de diseño en Python
- Python Cheatsheet
- venv Cheatsheet
- uv - Nuevo paquete, proyecto y gestor de entornos de Python
- Rendimiento de AWS Lambda: JavaScript vs Python vs Golang
- Construyendo microservicios orientados a eventos con AWS Kinesis