FastAPI y CQRS

26 de enero de 2026 Por: SiempreListo

cqrspythonfastapiapirest

Capítulo 18: FastAPI y CQRS

FastAPI es un framework web moderno para Python que destaca por su rendimiento (comparable a Node.js y Go) y su generación automática de documentación OpenAPI. Su diseño basado en type hints se integra naturalmente con CQRS.

Por qué FastAPI para CQRS

FastAPI ofrece características ideales para CQRS:

async nativo: Maneja miles de requests concurrentes
Pydantic: Validación automática de entrada con tipos
OpenAPI: Documentación generada automáticamente
Dependency Injection: Sistema de inyección de dependencias integrado

Configuración Inicial

El lifespan es un context manager que gestiona el ciclo de vida de la aplicación: ejecuta código al iniciar (yield anterior) y al cerrar (yield posterior).

app.state es un objeto para almacenar estado compartido entre requests, como los buses de comandos y queries.

@asynccontextmanager convierte una función generadora async en un context manager.

# src/main.py
from fastapi import FastAPI
from contextlib import asynccontextmanager
from .bootstrap import setup_command_bus, setup_query_bus, setup_dependencies

@asynccontextmanager
async def lifespan(app: FastAPI):
    deps = await setup_dependencies()
    app.state.command_bus = setup_command_bus(deps)
    app.state.query_bus = setup_query_bus(deps)
    yield
    await deps.close()

app = FastAPI(title="OrderFlow API", lifespan=lifespan)

Esquemas de Request/Response

Pydantic es una librería de validación de datos que usa type hints. BaseModel es la clase base para definir esquemas de datos con validación automática.

Field permite agregar restricciones de validación: ... indica que el campo es requerido, min_length=1 valida longitud mínima, gt=0 valida que sea mayor que cero.

Estos esquemas sirven como contrato de la API: FastAPI rechaza automáticamente requests que no cumplan las validaciones.

# src/api/schemas.py
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime

class CreateOrderRequest(BaseModel):
    customer_id: str = Field(..., min_length=1)

class AddItemRequest(BaseModel):
    product_id: str
    name: str
    price: float = Field(..., gt=0)
    quantity: int = Field(..., gt=0)

class OrderItemResponse(BaseModel):
    product_id: str
    name: str
    price: float
    quantity: int

class OrderResponse(BaseModel):
    id: str
    customer_id: str
    items: List[OrderItemResponse]
    status: str
    total: float
    created_at: datetime

class OrderListResponse(BaseModel):
    orders: List[OrderResponse]
    total: int
    page: int

Endpoints de Comandos

Los routers agrupan endpoints relacionados bajo un prefijo comun (/orders) y tags para la documentación.

status_code=status.HTTP_201_CREATED define el código de respuesta por defecto. 204 No Content se usa cuando el comando no retorna datos.

HTTPException es la forma de FastAPI de retornar errores HTTP con código y detalle.

# src/api/routes/commands.py
from fastapi import APIRouter, Request, HTTPException, status
from uuid import uuid4
from ..schemas import CreateOrderRequest, AddItemRequest
from ...command.handlers.create_order import CreateOrderCommand
from ...command.handlers.add_item import AddItemCommand

router = APIRouter(prefix="/orders", tags=["commands"])

@router.post("/", status_code=status.HTTP_201_CREATED)
async def create_order(request: Request, body: CreateOrderRequest):
    order_id = str(uuid4())
    command = CreateOrderCommand(order_id=order_id, customer_id=body.customer_id)

    try:
        await request.app.state.command_bus.dispatch(command)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

    return {"order_id": order_id}

@router.post("/{order_id}/items", status_code=status.HTTP_204_NO_CONTENT)
async def add_item(request: Request, order_id: str, body: AddItemRequest):
    command = AddItemCommand(
        order_id=order_id,
        product_id=body.product_id,
        name=body.name,
        price=body.price,
        quantity=body.quantity
    )

    try:
        await request.app.state.command_bus.dispatch(command)
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

Endpoints de Queries

response_model indica el tipo de respuesta para validación de salida y documentación OpenAPI.

Query de FastAPI define parámetros de query string con valores por defecto y validaciones. ge=0 significa “mayor o igual que 0”, le=100 “menor o igual que 100”.

Los parámetros de ruta (como {order_id}) se extraen automáticamente de la URL.

# src/api/routes/queries.py
from fastapi import APIRouter, Request, HTTPException, Query
from ..schemas import OrderResponse, OrderListResponse
from ...query.handlers.get_order import GetOrderQuery
from ...query.handlers.list_orders import ListOrdersQuery

router = APIRouter(prefix="/orders", tags=["queries"])

@router.get("/{order_id}", response_model=OrderResponse)
async def get_order(request: Request, order_id: str):
    query = GetOrderQuery(order_id=order_id)
    result = await request.app.state.query_bus.ask(query)

    if not result:
        raise HTTPException(status_code=404, detail="Order not found")
    return result

@router.get("/", response_model=OrderListResponse)
async def list_orders(
    request: Request,
    customer_id: str,
    page: int = Query(default=0, ge=0),
    limit: int = Query(default=20, ge=1, le=100)
):
    query = ListOrdersQuery(customer_id=customer_id, page=page, limit=limit)
    return await request.app.state.query_bus.ask(query)

Dependencias con FastAPI

Depends es el sistema de inyección de dependencias de FastAPI. Las funciones de dependencia se ejecutan antes del handler y sus resultados se inyectan como parámetros.

Este patrón es más limpio que acceder a request.app.state directamente en cada endpoint.

# src/api/dependencies.py
from fastapi import Request, Depends
from ..command.bus import CommandBus
from ..query.bus import QueryBus

def get_command_bus(request: Request) -> CommandBus:
    return request.app.state.command_bus

def get_query_bus(request: Request) -> QueryBus:
    return request.app.state.query_bus

# Uso alternativo con Depends
@router.post("/")
async def create_order(
    body: CreateOrderRequest,
    command_bus: CommandBus = Depends(get_command_bus)
):
    command = CreateOrderCommand(order_id=str(uuid4()), customer_id=body.customer_id)
    await command_bus.dispatch(command)
    return {"order_id": command.order_id}

Middleware de Errores

Los exception handlers capturan excepciones específicas globalmente y las convierten en respuestas HTTP consistentes.

JSONResponse permite construir respuestas HTTP con control total sobre status code y contenido.

Esto centraliza el manejo de errores: los handlers pueden lanzar excepciones de dominio sin preocuparse por HTTP.

# src/api/middleware.py
from fastapi import Request
from fastapi.responses import JSONResponse
from ..domain.exceptions import DomainException, OrderNotFoundError

async def domain_exception_handler(request: Request, exc: DomainException):
    return JSONResponse(
        status_code=400,
        content={"error": exc.__class__.__name__, "message": str(exc)}
    )

async def not_found_handler(request: Request, exc: OrderNotFoundError):
    return JSONResponse(
        status_code=404,
        content={"error": "NotFound", "message": str(exc)}
    )

# En main.py
app.add_exception_handler(DomainException, domain_exception_handler)
app.add_exception_handler(OrderNotFoundError, not_found_handler)

Registro de Rutas

include_router agrega un router al app principal. Cada router mantiene sus propios prefijos y configuraciones.

# src/main.py
from .api.routes import commands, queries

app.include_router(commands.router)
app.include_router(queries.router)

Resumen

FastAPI con CQRS ofrece:

Validación automática con Pydantic
Documentación OpenAPI generada
Async nativo para operaciones I/O
Inyección de dependencias flexible

Glosario

FastAPI

Definición: Framework web moderno para Python basado en Starlette (ASGI) y Pydantic, diseñado para crear APIs con alto rendimiento y documentación automática.

Por qué es importante: Combina velocidad de desarrollo con rendimiento de producción. La validación y documentación automáticas reducen errores y tiempo de desarrollo.

Ejemplo práctico: Un endpoint con def create_order(body: CreateOrderRequest) automáticamente valida el body, genera documentación OpenAPI y rechaza requests inválidos.

Pydantic

Definición: Librería de Python para validación de datos usando type hints. BaseModel es la clase base para definir esquemas de datos con validación automática.

Por qué es importante: Valida datos de entrada automáticamente, serializa/deserializa JSON, y genera esquemas JSON Schema para documentación.

Ejemplo práctico: price: float = Field(gt=0) valida que price sea un número mayor que cero. Si llega -5, Pydantic rechaza el request con un error detallado.

Lifespan

Definición: Context manager que gestiona el ciclo de vida de una aplicación FastAPI, ejecutando código al iniciar y al cerrar.

Por qué es importante: Permite inicializar recursos (conexiones DB, buses) al arrancar y liberarlos al cerrar, evitando memory leaks y conexiones huérfanas.

Ejemplo práctico: En lifespan se conecta a la base de datos al iniciar y se cierra la conexión al apagar el servidor, garantizando cleanup correcto.

APIRouter

Definición: Clase de FastAPI que agrupa endpoints relacionados con prefijos y configuraciones comunes.

Por qué es importante: Organiza endpoints en módulos separados, facilita testing y mantiene el archivo main.py limpio.

Ejemplo práctico: APIRouter(prefix="/orders", tags=["orders"]) agrupa todos los endpoints de pedidos bajo /orders y los documenta bajo el tag “orders” en OpenAPI.

Depends

Definición: Sistema de inyección de dependencias de FastAPI. Las funciones de dependencia se ejecutan antes del handler y sus resultados se pasan como parámetros.

Por qué es importante: Desacopla la obtención de dependencias del handler. Facilita testing (inyectar mocks) y reutilización de lógica común.

Ejemplo práctico: command_bus: CommandBus = Depends(get_command_bus) obtiene el bus sin que el handler sepa cómo se obtiene.

HTTPException

Definición: Excepción de FastAPI que se convierte automáticamente en una respuesta HTTP con el status code y detalle especificados.

Por qué es importante: Simplifica el manejo de errores. Lanzar HTTPException(404, "Not found") retorna una respuesta 404 formateada correctamente.

Ejemplo práctico: Si el pedido no existe, raise HTTPException(status_code=404, detail="Order not found") retorna {"detail": "Order not found"} con status 404.

Query (Parámetro)

Definición: Función de FastAPI para definir parámetros de query string con validaciones, valores por defecto y documentación.

Por qué es importante: Valida parámetros de URL automáticamente y los documenta en OpenAPI con restricciones y ejemplos.

Ejemplo práctico: page: int = Query(default=0, ge=0) acepta ?page=2 en la URL, valida que sea >= 0, y usa 0 si no se proporciona.

Exception Handler

Definición: Función registrada para capturar excepciones específicas globalmente y convertirlas en respuestas HTTP.

Por qué es importante: Centraliza el manejo de errores. Los handlers pueden lanzar excepciones de dominio sin saber nada de HTTP.

Ejemplo práctico: app.add_exception_handler(OrderNotFoundError, handler) captura todas las OrderNotFoundError y las convierte en respuestas 404 consistentes.

OpenAPI/Swagger

Definición: Especificación para describir APIs REST. FastAPI genera automáticamente documentación OpenAPI accesible en /docs (Swagger UI) y /redoc.

Por qué es importante: Documenta la API automáticamente basándose en tipos y schemas. Permite probar endpoints directamente desde el navegador.

Ejemplo práctico: Visitar /docs muestra todos los endpoints con sus parámetros, schemas de request/response, y permite ejecutar requests de prueba.

← Capítulo 17: Python CQRS | Capítulo 19: API REST →