Arquitectura Limpia en Python
Intermedio⏱70 min📖4 min de lecturaLección 6 de 7
Arquitectura Limpia en Python
Aplica los principios de la Arquitectura Limpia con herramientas propias de Python: Protocols, dataclasses, inyección de dependencias y estructura de proyecto
Python es un lenguaje excelente para la Arquitectura Limpia. Su tipado dinámico, las clases Protocol y su sintaxis sencilla facilitan definir límites, invertir dependencias y mantener pura la lógica de negocio.
Nota
La filosofía de Python encaja de forma natural con la Arquitectura Limpia: simplicidad, legibilidad y practicidad. No necesitas un contenedor de DI pesado: el duck typing y las dataclasses de Python bastan.
Estructura del proyecto
Un proyecto Python de Arquitectura Limpia sigue una estructura de directorios consistente:
my_project/
src/
entities/ # Enterprise business rules
__init__.py
order.py
customer.py
product.py
use_cases/ # Application business rules
__init__.py
place_order.py
cancel_order.py
interfaces.py # Repository & gateway protocols
interface_adapters/ # Adapters
__init__.py
controllers/
order_controller.py
presenters/
order_presenter.py
repositories/
postgres_order_repo.py
in_memory_order_repo.py
gateways/
stripe_payment.py
frameworks/ # Framework setup
__init__.py
web/
flask_app.py
fastapi_app.py
database/
postgres_connection.py
config.py
main.py # Composition root
tests/
test_entities/
test_use_cases/
test_adapters/
requirements.txt
pyproject.toml
Consejo
Mantén la estructura de directorios plana y descriptiva. Cada carpeta corresponde a una capa de la Arquitectura Limpia. El archivo interfaces.py de la capa de casos de uso define los límites que implementan las capas externas.
Uso de Protocol para los límites
El typing.Protocol de Python (PEP 544) ofrece subtipado estructural: no necesitas herencia explícita:
python
from typing import Protocol, runtime_checkable
# Define the protocol (interface) in the inner layer
@runtime_checkable
class OrderRepository(Protocol):
def save(self, order: "Order") -> None:
...
def find_by_id(self, order_id: str) -> "Order | None":
...
def find_by_customer(self, customer_id: str) -> list["Order"]:
...
# The use case depends on the Protocol
class CancelOrderUseCase:
def __init__(self, repo: OrderRepository):
self._repo = repo
def execute(self, order_id: str) -> None:
order = self._repo.find_by_id(order_id)
if order is None:
raise ValueError("Order not found")
order.cancel()
self._repo.save(order)
# Outer layer implementation — no explicit inheritance
class PostgresOrderRepository:
def save(self, order: "Order") -> None:
...
def find_by_id(self, order_id: str) -> "Order | None":
...
def find_by_customer(self, customer_id: str) -> list["Order"]:
...
# Testing — also conforms to the Protocol
class FakeOrderRepository:
def __init__(self):
self._orders: dict[str, "Order"] = {}
def save(self, order: "Order") -> None:
self._orders[order.order_id] = order
def find_by_id(self, order_id: str) -> "Order | None":
return self._orders.get(order_id)
def find_by_customer(self, customer_id: str) -> list["Order"]:
return [o for o in self._orders.values() if o.customer.customer_id == customer_id]Nota
A diferencia de las ABC, los Protocol usan el subtipado estructural: un objeto se considera que implementa el protocolo si tiene los métodos adecuados, independientemente de su jerarquía de clases. Es más pythónico y más flexible.
Dataclasses para DTOs y entidades
Las dataclasses son ideales para entidades y objetos de transferencia de datos:
python
from dataclasses import dataclass, field
from decimal import Decimal
from enum import Enum, auto
from typing import List, Optional
from datetime import datetime
class OrderStatus(Enum):
PENDING = auto()
CONFIRMED = auto()
SHIPPED = auto()
DELIVERED = auto()
CANCELLED = auto()
@dataclass
class Address:
street: str
city: str
state: str
zip_code: str
country: str
def full_address(self) -> str:
return f"{self.street}, {self.city}, {self.state} {self.zip_code}, {self.country}"
@dataclass
class Customer:
customer_id: str
name: str
email: str
address: Optional[Address] = None
@dataclass
class OrderItem:
product_id: str
product_name: str
quantity: int
unit_price: Decimal
def total(self) -> Decimal:
return Decimal(str(self.quantity)) * self.unit_price
@dataclass
class Order:
order_id: str
customer: Customer
items: List[OrderItem] = field(default_factory=list)
status: OrderStatus = OrderStatus.PENDING
created_at: datetime = field(default_factory=datetime.now)
_total_cache: Optional[Decimal] = field(default=None, repr=False)
def add_item(self, item: OrderItem) -> None:
if self.status != OrderStatus.PENDING:
raise ValueError("Cannot modify a non-pending order")
self.items.append(item)
self._total_cache = None
def remove_item(self, product_id: str) -> None:
if self.status != OrderStatus.PENDING:
raise ValueError("Cannot modify a non-pending order")
self.items = [i for i in self.items if i.product_id != product_id]
self._total_cache = None
def total(self) -> Decimal:
if self._total_cache is None:
self._total_cache = sum(
(item.total() for item in self.items),
Decimal("0"),
)
return self._total_cache
def confirm(self) -> None:
if self.status != OrderStatus.PENDING:
raise ValueError(f"Cannot confirm order in {self.status.name} state")
if not self.items:
raise ValueError("Cannot confirm an empty order")
self.status = OrderStatus.CONFIRMED
def cancel(self) -> None:
if self.status in (OrderStatus.SHIPPED, OrderStatus.DELIVERED):
raise ValueError(f"Cannot cancel order in {self.status.name} state")
self.status = OrderStatus.CANCELLED
# Input/Output DTOs
@dataclass
class PlaceOrderInput:
customer_id: str
items: List[dict]
coupon_code: Optional[str] = None
@dataclass
class PlaceOrderOutput:
order_id: str
total: float
status: str
item_count: int
created_at: strInyección de dependencias sin framework
Python no necesita un framework de DI: la inyección manual es explícita y limpia:
python
# --- composition_root.py ---
# This is the only place where concrete classes are instantiated
def create_place_order_use_case() -> "PlaceOrderUseCase":
order_repo = PostgresOrderRepository(
connection_string="postgresql://localhost:5432/shop"
)
customer_repo = PostgresCustomerRepository(
connection_string="postgresql://localhost:5432/shop"
)
product_repo = PostgresProductRepository(
connection_string="postgresql://localhost:5432/shop"
)
payment_gateway = StripePaymentGateway(api_key="sk_test_...")
notification = EmailNotificationService(smtp_host="smtp.example.com")
return PlaceOrderUseCase(
customer_repo=customer_repo,
product_repo=product_repo,
order_repo=order_repo,
payment_gateway=payment_gateway,
notification_service=notification,
)
def create_test_place_order_use_case() -> "PlaceOrderUseCase":
return PlaceOrderUseCase(
customer_repo=InMemoryCustomerRepository(),
product_repo=InMemoryProductRepository(),
order_repo=InMemoryOrderRepository(),
payment_gateway=FakePaymentGateway(),
notification_service=FakeNotificationService(),
)
# --- main.py ---
def main() -> None:
use_case = create_place_order_use_case()
controller = OrderController(use_case=use_case)
app = create_flask_app(controller=controller)
app.run(host="0.0.0.0", port=8000)
if __name__ == "__main__":
main()El caso de uso con límites completos
Este es un caso de uso completo y listo para producción:
python
from dataclasses import dataclass
from decimal import Decimal
from typing import Protocol, Optional
import uuid
from datetime import datetime
# --- Protocols (defined in use_cases/) ---
class CustomerRepository(Protocol):
def find_by_id(self, customer_id: str) -> Optional["Customer"]: ...
def save(self, customer: "Customer") -> None: ...
class ProductRepository(Protocol):
def find_by_id(self, product_id: str) -> Optional["Product"]: ...
class OrderRepository(Protocol):
def save(self, order: "Order") -> None: ...
def find_by_id(self, order_id: str) -> Optional["Order"]: ...
class PaymentGateway(Protocol):
def charge(self, customer_email: str, amount: Decimal) -> str: ...
def refund(self, transaction_id: str) -> Decimal: ...
class NotificationService(Protocol):
def send_order_confirmation(self, email: str, order_id: str) -> None: ...
# --- Input / Output DTOs ---
@dataclass
class PlaceOrderInput:
customer_id: str
items: list[dict]
coupon_code: Optional[str] = None
@dataclass
class PlaceOrderOutput:
order_id: str
total: float
status: str
item_count: int
transaction_id: str
created_at: str
# --- The Use Case ---
class PlaceOrderUseCase:
def __init__(
self,
customer_repo: CustomerRepository,
product_repo: ProductRepository,
order_repo: OrderRepository,
payment_gateway: PaymentGateway,
notification_service: NotificationService,
):
self._customer_repo = customer_repo
self._product_repo = product_repo
self._order_repo = order_repo
self._payment = payment_gateway
self._notification = notification_service
def execute(self, input_dto: PlaceOrderInput) -> PlaceOrderOutput:
customer = self._customer_repo.find_by_id(input_dto.customer_id)
if customer is None:
raise ValueError(f"Customer '{input_dto.customer_id}' not found")
order_id = str(uuid.uuid4())
order = Order(order_id=order_id, customer=customer)
for item_data in input_dto.items:
product = self._product_repo.find_by_id(item_data["product_id"])
if product is None:
raise ValueError(f"Product '{item_data['product_id']}' not found")
item = OrderItem(
product_id=product.product_id,
product_name=product.name,
quantity=item_data["quantity"],
unit_price=product.price,
)
order.add_item(item)
order.confirm()
total = order.total()
txn_id = self._payment.charge(customer.email, total)
self._order_repo.save(order)
self._notification.send_order_confirmation(customer.email, order_id)
return PlaceOrderOutput(
order_id=order_id,
total=float(total),
status=order.status.name,
item_count=len(order.items),
transaction_id=txn_id,
created_at=order.created_at.isoformat(),
)La capa de adaptadores en Python
python
# --- controllers/order_controller.py ---
import json
class OrderController:
def __init__(self, place_order_use_case: "PlaceOrderUseCase"):
self._use_case = place_order_use_case
def handle_place_order(self, request: dict) -> dict:
try:
body = self._parse_body(request)
self._validate_body(body)
input_dto = PlaceOrderInput(
customer_id=body["customer_id"],
items=body["items"],
coupon_code=body.get("coupon_code"),
)
output = self._use_case.execute(input_dto)
return {
"status": 201,
"body": {
"success": True,
"data": {
"order_id": output.order_id,
"total": output.total,
"status": output.status,
"item_count": output.item_count,
"transaction_id": output.transaction_id,
"created_at": output.created_at,
},
},
}
except ValueError as e:
return {"status": 400, "body": {"success": False, "error": str(e)}}
except Exception:
return {"status": 500, "body": {"success": False, "error": "Internal error"}}
def _parse_body(self, request: dict) -> dict:
raw = request.get("body", "{}")
if isinstance(raw, str):
return json.loads(raw)
return raw
def _validate_body(self, body: dict) -> None:
if "customer_id" not in body:
raise ValueError("Field 'customer_id' is required")
if "items" not in body or not isinstance(body["items"], list) or len(body["items"]) == 0:
raise ValueError("Field 'items' must be a non-empty list")
for item in body["items"]:
if "product_id" not in item:
raise ValueError("Each item must have 'product_id'")
if "quantity" not in item:
raise ValueError("Each item must have 'quantity'")Patrones de pruebas en Python
python
import pytest
from decimal import Decimal
# --- Fixtures ---
@pytest.fixture
def customer() -> Customer:
return Customer(
customer_id="CUST-001",
name="Alice Johnson",
email="alice@example.com",
)
@pytest.fixture
def sample_items() -> list[OrderItem]:
return [
OrderItem(product_id="P1", product_name="Widget", quantity=2, unit_price=Decimal("10.00")),
OrderItem(product_id="P2", product_name="Gadget", quantity=1, unit_price=Decimal("25.00")),
]
@pytest.fixture
def order(customer: Customer, sample_items: list[OrderItem]) -> Order:
return Order(
order_id="ORD-001",
customer=customer,
items=sample_items,
)
# --- Entity Tests ---
class TestOrderEntity:
def test_calculate_total(self, order: Order):
assert order.total() == Decimal("45.00")
def test_confirm_sets_status(self, order: Order):
order.confirm()
assert order.status == OrderStatus.CONFIRMED
def test_confirm_empty_order_fails(self, customer: Customer):
empty_order = Order(order_id="ORD-002", customer=customer)
with pytest.raises(ValueError, match="empty"):
empty_order.confirm()
def test_cancel_pending_order(self, order: Order):
order.cancel()
assert order.status == OrderStatus.CANCELLED
def test_cancel_shipped_order_fails(self, order: Order):
order.status = OrderStatus.SHIPPED
with pytest.raises(ValueError, match="cancel"):
order.cancel()
def test_add_item_to_confirmed_order_fails(self, order: Order):
order.confirm()
with pytest.raises(ValueError, match="pending"):
order.add_item(OrderItem(product_id="P3", product_name="X", quantity=1, unit_price=Decimal("1")))
# --- Use Case Tests ---
class TestPlaceOrderUseCase:
@pytest.fixture
def use_case(self):
customer_repo = InMemoryCustomerRepository()
customer_repo.save(Customer(customer_id="C1", name="Alice", email="a@test.com"))
product_repo = InMemoryProductRepository()
product_repo.save(Product(product_id="P1", name="Widget", price=Decimal("10.00"), stock=10))
product_repo.save(Product(product_id="P2", name="Gadget", price=Decimal("25.00"), stock=5))
return PlaceOrderUseCase(
customer_repo=customer_repo,
product_repo=product_repo,
order_repo=InMemoryOrderRepository(),
payment_gateway=FakePaymentGateway(),
notification_service=FakeNotificationService(),
)
def test_place_order_success(self, use_case):
output = use_case.execute(PlaceOrderInput(
customer_id="C1",
items=[{"product_id": "P1", "quantity": 2}, {"product_id": "P2", "quantity": 1}],
))
assert output.status == "CONFIRMED"
assert output.total == 45.0
assert output.item_count == 2
assert output.transaction_id.startswith("txn_")
def test_place_order_invalid_customer(self, use_case):
with pytest.raises(ValueError, match="not found"):
use_case.execute(PlaceOrderInput(customer_id="FAKE", items=[]))Patrones propios de Python
Administradores de contexto para los límites
python
from contextlib import contextmanager
from typing import Iterator
class UnitOfWork:
def __init__(self, connection_string: str):
self._conn_string = connection_string
@contextmanager
def transaction(self) -> Iterator["Connection"]:
import psycopg2
conn = psycopg2.connect(self._conn_string)
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
class PostgresOrderRepository:
def __init__(self, uow: UnitOfWork):
self._uow = uow
def save(self, order: "Order") -> None:
with self._uow.transaction() as conn:
with conn.cursor() as cur:
cur.execute("INSERT INTO orders ...", (...))Alias de tipos para mayor claridad
python
from typing import TypeAlias
from decimal import Decimal
JSON: TypeAlias = dict[str, "JSON"] | list["JSON"] | str | int | float | bool | None
OrderID: TypeAlias = str
CustomerID: TypeAlias = str
Money: TypeAlias = Decimal
Email: TypeAlias = str
class PaymentGateway(Protocol):
def charge(self, customer_email: Email, amount: Money) -> str: ...
def refund(self, transaction_id: str) -> Money: ...Excepciones personalizadas
python
class DomainError(Exception):
pass
class EntityNotFoundError(DomainError):
def __init__(self, entity_type: str, entity_id: str):
self.entity_type = entity_type
self.entity_id = entity_id
super().__init__(f"{entity_type} '{entity_id}' not found")
class InvalidOperationError(DomainError):
def __init__(self, operation: str, reason: str):
self.operation = operation
self.reason = reason
super().__init__(f"Cannot {operation}: {reason}")
# Usage in use case
class GetProductUseCase:
def __init__(self, repo: ProductRepository):
self._repo = repo
def execute(self, product_id: str) -> Product:
product = self._repo.find_by_id(product_id)
if product is None:
raise EntityNotFoundError("Product", product_id)
return productPydantic para la validación de DTOs
python
from pydantic import BaseModel, Field, field_validator
from typing import List, Optional
class PlaceOrderInput(BaseModel):
customer_id: str = Field(..., min_length=1)
items: List[dict]
coupon_code: Optional[str] = None
@field_validator("items")
@classmethod
def validate_items(cls, v: List[dict]) -> List[dict]:
if not v:
raise ValueError("At least one item required")
for item in v:
if "product_id" not in item:
raise ValueError("Each item needs product_id")
if "quantity" not in item or item["quantity"] < 1:
raise ValueError("Each item needs quantity >= 1")
return v
class PlaceOrderOutput(BaseModel):
order_id: str
total: float
status: str
item_count: int
transaction_id: str
created_at: strConsideraciones de rendimiento
Aspecto | Solución en Python | Impacto |
|---|---|---|
Costo de los límites | Llamadas a Protocol | Mínimo: despacho dinámico de Python |
Contenedor de DI | Inyección manual | Cero overhead |
Creación de DTOs | Instanciación de dataclasses | Despreciable |
Patrón repositorio | SQL frente a ORM | El ORM añade un 10-30% de latencia |
Pruebas | Repositorios en memoria | 100 veces más rápido que la base de datos |
Advertencia
No sobreabstragas. Cada límite tiene un costo. Empieza con dependencias directas y extrae interfaces solo cuando las necesites. La abstracción prematura es tan dañina como la ausencia de abstracción.
Generador completo de esqueleto de proyecto
python
#!/usr/bin/env python3
"""Generate a Clean Architecture Python project skeleton."""
import os
import sys
def create_project(name: str) -> None:
directories = [
f"src/entities",
f"src/use_cases",
f"src/interface_adapters/controllers",
f"src/interface_adapters/presenters",
f"src/interface_adapters/repositories",
f"src/interface_adapters/gateways",
f"src/frameworks/web",
f"src/frameworks/database",
f"tests/test_entities",
f"tests/test_use_cases",
f"tests/test_adapters",
]
for directory in directories:
os.makedirs(f"{name}/{directory}", exist_ok=True)
init_file = f"{name}/{directory}/__init__.py"
if not os.path.exists(init_file):
with open(init_file, "w") as f:
f.write("")
with open(f"{name}/src/main.py", "w") as f:
f.write('"""Composition root — wire dependencies here."""\n\n')
f.write("def create_app():\n")
f.write(' """Build and return the application."""\n')
f.write(" pass\n\n\n")
f.write('if __name__ == "__main__":\n')
f.write(" app = create_app()\n")
f.write(" app.run()\n")
with open(f"{name}/README.md", "w") as f:
f.write(f"# {name}\n\nClean Architecture project.\n")
print(f"Project '{name}' created successfully.")
print(f"cd {name} && python src/main.py")
if __name__ == "__main__":
if len(sys.argv) != 2:
print("Usage: python generate_skeleton.py <project_name>")
sys.exit(1)
create_project(sys.argv[1])Comparación: Python frente a otros lenguajes
Aspecto | Python | Java / C# | TypeScript |
|---|---|---|---|
Interfaces | Protocol / ABC | palabra interface | interface |
DI | Manual / librerías sencillas | Spring / Guice | InversifyJS |
Entidades | @dataclass | POJO / record | class / type |
DTOs | @dataclass / Pydantic | record | interface |
Pruebas | pytest + in-memory | JUnit + mocks | Jest + mocks |
Compilación | Sin paso de compilación | Maven / Gradle | tsc / esbuild |
Ejercicios prácticos
-
Límite con Protocol: Define un Protocol
Loggerconlog(level, message). ImplementaStdoutLoggeryFileLogger. Escribe un caso de uso que lo utilice. -
DTOs como dataclasses: Crea dataclasses de entrada/salida para un
TransferMoneyUseCase. Incluyefrom_account_id,to_account_id,amountycurrency. -
DI manual: Escribe una raíz de composición que cree un
ProcessOrderUseCaseconPostgresOrderRepository,StripePaymentGatewayyEmailNotificationService. -
Repositorios en memoria: Implementa
InMemoryAccountRepositoryconsave,find_by_idyfind_by_owner. Prueba un caso de uso que transfiere dinero entre dos cuentas. -
Generador de proyecto: Ejecuta el generador de esqueletos de arriba. Añade dos casos de uso (
CreateUseryDeleteUser) con sus repositorios y controladores correspondientes. -
Refactoriza a Protocol: Toma una clase que hereda de
ABCy refactorízala para usarProtocol. Muestra ambas versiones. -
Validación con Pydantic: Sustituye un DTO dataclass normal por un modelo de Pydantic que valide el formato del correo y precios positivos.
-
Patrón Unit of Work: Implementa un administrador de contexto
UnitOfWorky úsalo en un repositorio. Escribe una prueba que verifique el rollback cuando algo falla.
Éxito
Python y la Arquitectura Limpia encajan de maravilla. Los Protocol, las dataclasses y la simplicidad de Python te permiten concentrarte en lo que importa: una lógica de negocio limpia y fácil de probar.
Progreso86%