Clean Architecture con FastAPI

Python · AWS · Arquitectura

Cómo separar tu dominio de la infraestructura en proyectos FastAPI reales, con ejemplos de integración con AWS Lambda y DynamoDB.

Un recuerdo de 2008

Hace 16 años escribí un script en PHP que leía archivos de texto, formaba grupos aleatorios de alumnos y pintaba tablas HTML — todo en menos de 30 minutos y en un solo archivo. Funcionaba. Pero era imposible de testear, reutilizar o cambiar sin romper algo. En ese entonces no existía el vocabulario para describirlo — hoy lo llamaríamos una violación total de separación de responsabilidades.

— Ejercicio PHP: Grupos aleatorios de alumnos (2008)

2008 · PHP	Hoy · FastAPI
Lógica, datos y HTML mezclados en un solo bloque. Rápido de escribir, imposible de mantener.	Dominio, aplicación, infraestructura y entrypoints como capas independientes.

Ese error — mezclar la lógica de negocio con los detalles de infraestructura — lo seguí cometiendo años después, ya con frameworks modernos. Rutas que hacen queries directamente a la base de datos, funciones que dependen de boto3 hardcodeado, tests imposibles de escribir sin levantar DynamoDB local.

La señal de alarma es simple: si no puedes probar tu lógica de negocio sin tocar la base de datos, algo está mal.

Clean Architecture resuelve exactamente esto: tus reglas de negocio no saben nada de FastAPI, DynamoDB, ni de ningún framework externo. Solo saben de tu dominio.

Las capas y qué va en cada una

┌─────────────────────────────────────────────────────────────┐
│  Dominio        Entidades · Reglas de negocio · Interfaces  │
├─────────────────────────────────────────────────────────────┤
│  Aplicación     Casos de uso · DTOs · Servicios de dominio  │
├─────────────────────────────────────────────────────────────┤
│  Infraestructura    DynamoDB · S3 · Implementaciones        │
├─────────────────────────────────────────────────────────────┤
│  Entrypoints    FastAPI routes · Lambda handlers · CLI      │
└─────────────────────────────────────────────────────────────┘

La regla central: las flechas de dependencia apuntan siempre hacia adentro. El dominio no importa nada de las capas externas.

Estructura del proyecto

src/
  domain/
    entities.py        # Pydantic models puros
    repositories.py    # Interfaces (ABCs)
    exceptions.py
  application/
    use_cases.py       # Lógica orquestadora
    dtos.py
  infrastructure/
    dynamo_repo.py     # Implementación real
    memory_repo.py     # Implementación para tests
  entrypoints/
    api/               # FastAPI routers
    lambda_handler.py

El dominio: sin frameworks

Las entidades son modelos de datos puros. El repositorio es una interfaz abstracta que el dominio define pero no implementa.

# domain/entities.py
from dataclasses import dataclass
from datetime import datetime

@dataclass
class Order:
    id: str
    user_id: str
    total: float
    status: str
    created_at: datetime

    def can_be_cancelled(self) -> bool:
        return self.status in ("pending", "processing")


# domain/repositories.py
from abc import ABC, abstractmethod
from .entities import Order

class OrderRepository(ABC):
    @abstractmethod
    async def get_by_id(self, order_id: str) -> Order | None: ...

    @abstractmethod
    async def save(self, order: Order) -> None: ...

Caso de uso: la lógica orquestadora

El caso de uso recibe el repositorio como dependencia. No sabe si es DynamoDB o un dict en memoria.

# application/use_cases.py
from domain.repositories import OrderRepository
from domain.exceptions import OrderNotFound, CannotCancelOrder

class CancelOrderUseCase:
    def __init__(self, repo: OrderRepository):
        self.repo = repo

    async def execute(self, order_id: str) -> None:
        order = await self.repo.get_by_id(order_id)

        if order is None:
            raise OrderNotFound(order_id)

        if not order.can_be_cancelled():
            raise CannotCancelOrder(order.status)

        order.status = "cancelled"
        await self.repo.save(order)

Este caso de uso es testeable con un simple InMemoryOrderRepository. No necesitas mocks complicados ni DynamoDB local.

Infraestructura: DynamoDB

La implementación concreta sabe cómo hablar con DynamoDB. Implementa la interfaz del dominio.

# infrastructure/dynamo_repo.py
import boto3
from domain.repositories import OrderRepository
from domain.entities import Order

class DynamoOrderRepository(OrderRepository):
    def __init__(self, table_name: str):
        self.table = boto3.resource("dynamodb").Table(table_name)

    async def get_by_id(self, order_id: str) -> Order | None:
        resp = self.table.get_item(Key={"id": order_id})
        item = resp.get("Item")
        if not item:
            return None
        return Order(**item)

    async def save(self, order: Order) -> None:
        self.table.put_item(Item=vars(order))

Entrypoint: FastAPI + inyección de dependencias

FastAPI conecta todo con su sistema de Depends. El router no sabe nada de DynamoDB, solo recibe el caso de uso ya configurado.

# entrypoints/api/orders.py
from fastapi import APIRouter, Depends, HTTPException
from application.use_cases import CancelOrderUseCase
from infrastructure.dynamo_repo import DynamoOrderRepository
from domain.exceptions import OrderNotFound, CannotCancelOrder
import os

router = APIRouter(prefix="/orders")

def get_cancel_use_case() -> CancelOrderUseCase:
    repo = DynamoOrderRepository(os.environ["ORDERS_TABLE"])
    return CancelOrderUseCase(repo)

@router.delete("/{order_id}")
async def cancel_order(
    order_id: str,
    use_case: CancelOrderUseCase = Depends(get_cancel_use_case),
):
    try:
        await use_case.execute(order_id)
        return {"status": "cancelled"}
    except OrderNotFound:
        raise HTTPException(404, "Order not found")
    except CannotCancelOrder as e:
        raise HTTPException(422, str(e))

El mismo handler para AWS Lambda

Aquí está la verdadera ventaja: el mismo caso de uso funciona dentro de una Lambda sin tocar una sola línea de lógica de negocio.

# entrypoints/lambda_handler.py
from mangum import Mangum
from fastapi import FastAPI
from entrypoints.api.orders import router

app = FastAPI()
app.include_router(router)

handler = Mangum(app)  # AWS Lambda entry point

Mangum traduce los eventos de API Gateway al formato ASGI que FastAPI entiende. Sin cambiar nada de tu dominio o aplicación.

Testing: el premio final

# tests/test_cancel_order.py
import pytest
from domain.entities import Order
from application.use_cases import CancelOrderUseCase
from datetime import datetime

class InMemoryOrderRepository:
    def __init__(self, orders):
        self._orders = {o.id: o for o in orders}

    async def get_by_id(self, order_id):
        return self._orders.get(order_id)

    async def save(self, order):
        self._orders[order.id] = order

@pytest.mark.asyncio
async def test_cancel_pending_order():
    order = Order("1", "user-42", 99.0, "pending", datetime.now())
    repo = InMemoryOrderRepository([order])
    use_case = CancelOrderUseCase(repo)

    await use_case.execute("1")

    saved = await repo.get_by_id("1")
    assert saved.status == "cancelled"

Lo que ganaste

Tests sin base de datos
Cambiar DynamoDB por PostgreSQL sin tocar business logic
Lambda + FastAPI con el mismo código
Dominio legible por cualquier dev

En resumen

Clean Architecture no es burocracia: es la diferencia entre un script que funciona hoy y un sistema que puedes mantener en 5 años. El PHP de 2008 era válido para su contexto — rápido, directo, sin pretensiones. Pero si ese mismo código viviera en producción hoy, sería deuda técnica pura.

Con FastAPI el salto es especialmente natural porque su sistema de Depends fue diseñado exactamente para esto. Empieza pequeño: separa tus entidades en un domain/, define una interfaz de repositorio, mueve la lógica de tus rutas a un caso de uso. El resto llega solo.

Clean Architecture con FastAPI: Separando tu dominio de la infraestructura