Microservicios con FastAPI y Redis Streams: guía práctica en Python

Por qué FastAPI encaja tan bien en una arquitectura de microservicios

Dividir una aplicación en servicios pequeños e independientes ya no es una decisión exclusiva de las grandes empresas. Equipos medianos recurren a esta arquitectura para desplegar partes del sistema por separado, escalar solo lo que sufre carga y aislar fallos. El problema es que cada servicio adicional trae su propia API, su propio almacenamiento y la necesidad de hablar con los demás, así que el framework que elijas debe ser ligero, rápido de arrancar y cómodo para escribir contratos de datos claros.

FastAPI cumple esos requisitos. Se apoya en Starlette para la capa ASGI y en Pydantic para la validación, genera documentación OpenAPI automáticamente y permite mezclar funciones síncronas y asíncronas sin ceremonia. En este tutorial construirás un sistema de ejemplo formado por dos servicios: un catálogo de productos para una tienda de equipos de café y un servicio de pedidos que cobra, confirma y notifica. Para persistir datos y propagar eventos usarás Redis, que en su versión 8 ya incluye de serie el soporte de documentos JSON, el motor de búsqueda y, por supuesto, Redis Streams.

A lo largo del recorrido verás cómo modelar datos con Redis OM, cómo resolver la comunicación entre microservicios con FastAPI mediante peticiones HTTP asíncronas, cómo delegar trabajo lento a tareas en segundo plano y cómo desacoplar los servicios con eventos. Al final tendrás todo orquestado con Docker Compose y una lista de errores frecuentes con su solución.

La arquitectura que vamos a construir

Antes de escribir código conviene fijar las responsabilidades de cada pieza. Un microservicio bien delimitado es dueño de sus datos: nadie más escribe en su almacenamiento, y cualquier otro servicio que necesite esa información la pide por API o la recibe mediante eventos.

El sistema tendrá estos componentes:

• Servicio de catálogo: registra productos con nombre, precio y existencias. Expone un CRUD en el puerto 8000.
• Servicio de pedidos: recibe la intención de compra, consulta el precio vigente al catálogo, calcula comisiones y procesa el pago. Escucha en el puerto 8001.
• Redis 8: actúa como base de datos de ambos servicios (cada uno con su propio prefijo de claves) y como bus de eventos mediante streams.
• Workers: procesos independientes que consumen eventos. El del catálogo descuenta existencias cuando un pedido se completa; el de pedidos marca reembolsos cuando no hay stock suficiente.

El flujo de una compra queda así:

Cliente ──POST /pedidos──▶ Pedidos ──GET /productos/{id}──▶ Catálogo
                              │
                              │ (tarea en segundo plano: cobro simulado)
                              ▼
                     XADD pedidos:completados
                              │
                              ▼
                   Worker del catálogo ──descuenta stock──▶ Redis
                              │ (si no hay stock)
                              ▼
                     XADD pedidos:reembolsos
                              │
                              ▼
                   Worker de pedidos ──estado = reembolsado

La estructura del repositorio separa cada servicio en su propio directorio, con su Dockerfile y sus dependencias. Esto permite versionarlos juntos mientras aprendes, pero desplegarlos de forma independiente.

tienda-cafe/
├── compose.yaml
├── catalogo/
│   ├── Dockerfile
│   ├── requirements.txt
│   ├── config.py
│   ├── modelos.py
│   ├── main.py
│   └── worker.py
└── pedidos/
    ├── Dockerfile
    ├── requirements.txt
    ├── config.py
    ├── modelos.py
    ├── main.py
    ├── tareas.py
    └── worker.py

Versiones y preparación del entorno

Este tutorial está verificado con las versiones estables publicadas a la fecha de redacción: Python 3.15, FastAPI 0.162, Pydantic 2.14, Uvicorn 0.54, redis-py 8.1, Redis OM 1.2, HTTPX 0.28 y la imagen oficial de Redis 8.8. FastAPI sigue en la serie 0.x, lo que significa que una versión menor puede introducir cambios incompatibles; por eso es recomendable fijar las dependencias en un rango acotado.

Crea un entorno virtual por servicio. Durante el desarrollo puedes trabajar con uno solo, pero separarlos te obliga a declarar las dependencias reales de cada servicio y evita sorpresas al construir las imágenes.

mkdir -p tienda-cafe/catalogo tienda-cafe/pedidos
cd tienda-cafe/catalogo
python3.15 -m venv .venv
source .venv/bin/activate

El archivo de dependencias del catálogo es breve. El extra standard de FastAPI instala Uvicorn y el comando fastapi, que sustituye la vieja costumbre de invocar uvicorn main:app --reload a mano.

# catalogo/requirements.txt
fastapi[standard]>=0.162,<0.163
redis-om>=1.2,<1.3
redis>=8.1,<9

El servicio de pedidos añade HTTPX para llamar al catálogo de forma asíncrona. Muchos ejemplos antiguos usan requests dentro de los endpoints, lo cual bloquea el bucle de eventos; con HTTPX y async def el servidor puede atender otras peticiones mientras espera la respuesta.

# pedidos/requirements.txt
fastapi[standard]>=0.162,<0.163
redis-om>=1.2,<1.3
redis>=8.1,<9
httpx>=0.28,<0.29

pip install -r requirements.txt

Redis 8 como base de datos y bus de eventos

Hasta hace poco, para usar RedisJSON y RediSearch necesitabas la imagen redis/redis-stack o un servicio gestionado con esos módulos. Desde Redis 8, la distribución Open Source integra JSON, el motor de consultas, series temporales y estructuras probabilísticas en la imagen oficial. Para un entorno local basta con un contenedor:

docker run -d --name redis-tienda -p 6379:6379 redis:8.8
docker exec -it redis-tienda redis-cli PING

PONG

Si prefieres una interfaz gráfica para inspeccionar claves y streams, Redis Insight se distribuye como aplicación de escritorio y como imagen de contenedor independiente. No es obligatorio para seguir el tutorial, pero resulta muy útil para ver cómo se acumulan los mensajes en un stream y qué consumidores tienen mensajes pendientes.

Cada servicio leerá la URL de Redis desde una variable de entorno. Centralizar la configuración en un módulo pequeño evita repetir os.getenv por todo el código:

# catalogo/config.py
import os

from redis_om import get_redis_connection

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")

redis = get_redis_connection(url=REDIS_URL, decode_responses=True)

STREAM_COMPLETADOS = "pedidos:completados"
STREAM_REEMBOLSOS = "pedidos:reembolsos"

El parámetro decode_responses=True hace que redis-py devuelva cadenas en lugar de bytes, lo que simplifica mucho el manejo de los mensajes de los streams.

Modelado del catálogo con Redis OM

Redis OM es un mapeador de objetos que convierte modelos de Pydantic en hashes o documentos JSON de Redis. La versión 1.x está construida sobre Pydantic 2, de modo que la validación, los tipos y la serialización funcionan igual que en cualquier modelo de FastAPI.

Un HashModel almacena cada instancia como un hash plano, ideal para registros sencillos como un producto. Si necesitas listas o modelos anidados, JsonModel guarda el objeto como documento JSON.

# catalogo/modelos.py
from pydantic import BaseModel, Field, PositiveInt
from redis_om import HashModel

from config import redis


class Producto(HashModel):
    nombre: str
    precio: float
    existencias: int

    class Meta:
        database = redis
        global_key_prefix = "catalogo"


class ProductoEntrada(BaseModel):
    nombre: str = Field(min_length=3, max_length=80)
    precio: float = Field(gt=0)
    existencias: PositiveInt

Hay una decisión deliberada aquí: el modelo persistido (Producto) y el modelo de entrada (ProductoEntrada) son clases distintas. El primero refleja lo que guardas; el segundo define el contrato público de la API y sus restricciones. Separarlos evita que un cliente pueda enviar campos internos, como la clave primaria, y te deja evolucionar el almacenamiento sin romper a los consumidores.

El prefijo global_key_prefix hace que las claves queden como catalogo:modelos.Producto:<id>. Cuando dos servicios comparten una instancia de Redis, esta separación es lo mínimo exigible para que no se pisen.

Endpoints del servicio de catálogo

Con el modelo listo, el CRUD es directo. Los métodos de Redis OM son síncronos, así que los endpoints se declaran con def; FastAPI los ejecuta en un pool de hilos y el bucle de eventos queda libre.

# catalogo/main.py
from fastapi import FastAPI, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware
from redis_om import NotFoundError

from modelos import Producto, ProductoEntrada

app = FastAPI(title="Catálogo de equipos de café", version="1.0.0")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],
    allow_methods=["GET", "POST", "DELETE"],
    allow_headers=["*"],
)


def buscar_producto(pk: str) -> Producto:
    try:
        return Producto.get(pk)
    except NotFoundError:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "Producto no encontrado")


@app.get("/productos")
def listar_productos() -> list[dict]:
    return [buscar_producto(pk).model_dump() for pk in Producto.all_pks()]


@app.get("/productos/{pk}")
def obtener_producto(pk: str) -> dict:
    return buscar_producto(pk).model_dump()


@app.post("/productos", status_code=status.HTTP_201_CREATED)
def crear_producto(datos: ProductoEntrada) -> dict:
    producto = Producto(**datos.model_dump())
    producto.save()
    return producto.model_dump()


@app.delete("/productos/{pk}", status_code=status.HTTP_204_NO_CONTENT)
def eliminar_producto(pk: str) -> None:
    if not Producto.delete(pk):
        raise HTTPException(status.HTTP_404_NOT_FOUND, "Producto no encontrado")

Si migras código escrito con Pydantic 1, recuerda que .dict() está deprecado y ahora se usa .model_dump(), y que .json() se sustituye por .model_dump_json(). Arranca el servicio en modo desarrollo:

fastapi dev main.py --port 8000

Prueba a crear un producto y a listarlo:

curl -s -X POST http://localhost:8000/productos \
  -H "Content-Type: application/json" \
  -d '{"nombre": "Molino cónico manual", "precio": 89.9, "existencias": 12}'

{"pk":"01JBX3T8Z6Q4M2N7K5R9V0WXYZ","nombre":"Molino cónico manual","precio":89.9,"existencias":12}

La documentación interactiva queda disponible en http://localhost:8000/docs, generada a partir de las anotaciones de tipo. Es una de las razones por las que FastAPI resulta tan cómodo cuando varios equipos consumen tus servicios: el contrato está siempre sincronizado con el código.

El servicio de pedidos y las llamadas HTTP entre servicios

El servicio de pedidos necesita conocer el precio actual del producto. Podría copiar el catálogo en su propia base de datos, pero para este flujo una consulta síncrona por HTTP es suficiente y garantiza un precio vigente en el momento de la compra.

Primero, la configuración y el modelo:

# pedidos/config.py
import os

from redis_om import get_redis_connection

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
CATALOGO_URL = os.getenv("CATALOGO_URL", "http://localhost:8000")
COMISION = float(os.getenv("COMISION", "0.15"))

redis = get_redis_connection(url=REDIS_URL, decode_responses=True)

STREAM_COMPLETADOS = "pedidos:completados"
STREAM_REEMBOLSOS = "pedidos:reembolsos"

# pedidos/modelos.py
from pydantic import BaseModel, Field
from redis_om import HashModel

from config import redis


class Pedido(HashModel):
    producto_id: str
    cantidad: int
    precio_unitario: float
    comision: float
    total: float
    estado: str  # pendiente, completado, reembolsado

    class Meta:
        database = redis
        global_key_prefix = "pedidos"


class PedidoEntrada(BaseModel):
    producto_id: str
    cantidad: int = Field(ge=1, le=20)

Para las peticiones salientes conviene reutilizar un único httpx.AsyncClient durante toda la vida de la aplicación, porque mantiene un pool de conexiones y evita abrir un socket nuevo en cada pedido. La forma recomendada de gestionar recursos de arranque y apagado es el parámetro lifespan; los decoradores @app.on_event("startup") y @app.on_event("shutdown") que aparecen en tutoriales antiguos están deprecados.

# pedidos/main.py
from contextlib import asynccontextmanager

import httpx
from fastapi import BackgroundTasks, FastAPI, HTTPException, Request, status
from redis_om import NotFoundError

from config import CATALOGO_URL, COMISION
from modelos import Pedido, PedidoEntrada
from tareas import procesar_pago


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(base_url=CATALOGO_URL, timeout=3.0)
    yield
    await app.state.http.aclose()


app = FastAPI(title="Servicio de pedidos", version="1.0.0", lifespan=lifespan)


async def consultar_producto(http: httpx.AsyncClient, producto_id: str) -> dict:
    try:
        respuesta = await http.get(f"/productos/{producto_id}")
    except httpx.TransportError:
        raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, "Catálogo no disponible")

    if respuesta.status_code == status.HTTP_404_NOT_FOUND:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "El producto no existe")
    respuesta.raise_for_status()
    return respuesta.json()


@app.post("/pedidos", status_code=status.HTTP_202_ACCEPTED)
async def crear_pedido(
    datos: PedidoEntrada, request: Request, tareas: BackgroundTasks
) -> dict:
    producto = await consultar_producto(request.app.state.http, datos.producto_id)

    subtotal = producto["precio"] * datos.cantidad
    pedido = Pedido(
        producto_id=datos.producto_id,
        cantidad=datos.cantidad,
        precio_unitario=producto["precio"],
        comision=round(subtotal * COMISION, 2),
        total=round(subtotal * (1 + COMISION), 2),
        estado="pendiente",
    )
    pedido.save()

    tareas.add_task(procesar_pago, pedido.pk)
    return pedido.model_dump()


@app.get("/pedidos/{pk}")
def obtener_pedido(pk: str) -> dict:
    try:
        return Pedido.get(pk).model_dump()
    except NotFoundError:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "Pedido no encontrado")

Observa tres detalles. El cliente tiene un timeout explícito: sin él, un catálogo colgado bloquearía las peticiones de pedidos de forma indefinida. Los errores de transporte se traducen a un 503, un código que comunica al cliente que el problema es temporal. Y el endpoint responde con 202 Accepted, porque el pedido aún no está pagado; solo se ha aceptado para procesarlo.

Tareas en segundo plano para no bloquear al cliente

Cobrar con una pasarela de pago puede tardar varios segundos. Hacer esperar al cliente durante ese tiempo empeora la experiencia y consume conexiones del servidor. FastAPI ofrece BackgroundTasks, que ejecuta una función justo después de enviar la respuesta, dentro del mismo proceso.

# pedidos/tareas.py
import time

from config import STREAM_COMPLETADOS, redis
from modelos import Pedido


def procesar_pago(pedido_id: str) -> None:
    pedido = Pedido.get(pedido_id)

    time.sleep(4)  # simulación de la pasarela de pago

    pedido.estado = "completado"
    pedido.save()

    redis.xadd(
        STREAM_COMPLETADOS,
        {
            "pedido_id": pedido.pk,
            "producto_id": pedido.producto_id,
            "cantidad": pedido.cantidad,
        },
        maxlen=10_000,
        approximate=True,
    )

Como procesar_pago es una función síncrona, FastAPI la ejecuta en el pool de hilos y la llamada a time.sleep no congela el servidor. Si la escribieras con async def, tendrías que usar await asyncio.sleep(...), porque un time.sleep dentro de una corrutina detiene todo el bucle de eventos.

Las tareas en segundo plano de FastAPI son perfectas para trabajos breves y tolerantes a pérdidas, como enviar un correo o registrar métricas. Tienen una limitación importante: viven en la memoria del proceso. Si el contenedor se reinicia mientras la tarea duerme, el pedido queda en estado pendiente para siempre. Más adelante verás cómo mitigar ese riesgo.

El parámetro maxlen con approximate=True recorta el stream para que no crezca sin límite. Redis elimina las entradas más antiguas de forma eficiente, sin garantizar el número exacto, lo que es más barato que un recorte estricto.

Eventos con Redis Streams y grupos de consumidores

Ahora llega la parte que desacopla de verdad los servicios. El servicio de pedidos no llama al catálogo para descontar existencias: publica un hecho («este pedido se completó») y se desentiende. Quien esté interesado lo consume a su ritmo.

Un stream de Redis es un registro de solo escritura al final, con identificadores ordenados en el tiempo. A diferencia de Pub/Sub, los mensajes persisten: si el consumidor está caído cuando se publica el evento, lo leerá al volver. Los grupos de consumidores en Redis añaden dos garantías clave para microservicios: reparten los mensajes entre varias réplicas de un mismo worker y llevan la cuenta de qué mensajes se entregaron pero todavía no se confirmaron.

Los comandos fundamentales son pocos:

• XADD publica un mensaje.
• XGROUP CREATE crea un grupo de consumidores sobre un stream.
• XREADGROUP entrega mensajes nuevos a un consumidor concreto del grupo.
• XACK confirma que un mensaje se procesó.
• XAUTOCLAIM reasigna mensajes que llevan demasiado tiempo pendientes en un consumidor que murió.

Puedes experimentar desde redis-cli antes de escribir el worker:

redis-cli XGROUP CREATE pedidos:completados catalogo 0 MKSTREAM
redis-cli XADD pedidos:completados '*' pedido_id demo producto_id abc cantidad 1
redis-cli XREADGROUP GROUP catalogo consola COUNT 1 STREAMS pedidos:completados '>'

1) 1) "pedidos:completados"
   2) 1) 1) "1793082454321-0"
         2) 1) "pedido_id"
            2) "demo"
            3) "producto_id"
            4) "abc"
            5) "cantidad"
            6) "1"

El símbolo > significa «dame mensajes que nadie de este grupo haya recibido todavía». Hasta que ejecutes XACK, ese mensaje figura en la lista de pendientes del consumidor consola.

El worker del catálogo: descontar existencias de forma atómica

El worker es un proceso aparte que comparte modelos y configuración con la API del catálogo, pero no depende de ella para funcionar. Así puedes escalarlo o reiniciarlo sin tocar los endpoints.

# catalogo/worker.py
import logging
import socket

from redis.exceptions import ResponseError

from config import STREAM_COMPLETADOS, STREAM_REEMBOLSOS, redis
from modelos import Producto

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("worker-catalogo")

GRUPO = "catalogo"
CONSUMIDOR = socket.gethostname()


def asegurar_grupo() -> None:
    try:
        redis.xgroup_create(STREAM_COMPLETADOS, GRUPO, id="0", mkstream=True)
    except ResponseError as error:
        if "BUSYGROUP" not in str(error):
            raise


def descontar_existencias(evento: dict) -> None:
    clave = Producto.make_primary_key(evento["producto_id"])
    cantidad = int(evento["cantidad"])

    if not redis.exists(clave):
        log.warning("Producto %s inexistente, se reembolsa", evento["producto_id"])
        redis.xadd(STREAM_REEMBOLSOS, evento)
        return

    restantes = redis.hincrby(clave, "existencias", -cantidad)
    if restantes < 0:
        redis.hincrby(clave, "existencias", cantidad)
        redis.xadd(STREAM_REEMBOLSOS, evento)
        log.info("Sin stock para el pedido %s", evento["pedido_id"])


def ejecutar() -> None:
    asegurar_grupo()
    log.info("Consumidor %s escuchando %s", CONSUMIDOR, STREAM_COMPLETADOS)
    while True:
        lotes = redis.xreadgroup(
            GRUPO, CONSUMIDOR, {STREAM_COMPLETADOS: ">"}, count=20, block=5000
        )
        for _stream, mensajes in lotes or []:
            for mensaje_id, evento in mensajes:
                descontar_existencias(evento)
                redis.xack(STREAM_COMPLETADOS, GRUPO, mensaje_id)


if __name__ == "__main__":
    ejecutar()

El punto delicado es el descuento. La tentación es cargar el producto con Producto.get, restar en Python y llamar a save(). Funciona con un único worker, pero con dos réplicas procesando pedidos del mismo producto aparece una condición de carrera: ambas leen 5 unidades, ambas restan 1 y ambas guardan 4. Con HINCRBY la operación ocurre dentro de Redis de forma atómica, y el valor devuelto te dice si te has quedado en negativo para revertir y emitir el reembolso.

Para un control todavía más estricto puedes encapsular la comprobación y el descuento en un script Lua o una función de Redis, de modo que nunca exista, ni por un instante, un valor negativo visible para otros lectores.

Compensaciones: el worker de reembolsos

En un sistema distribuido no hay transacciones que abarquen varios servicios. Cuando un paso falla después de que otro ya tuvo éxito, la solución es una acción compensatoria. Aquí el cobro ya se hizo, pero no había existencias, así que el servicio de pedidos debe marcar el pedido como reembolsado (y, en un sistema real, devolver el dinero mediante la pasarela).

# pedidos/worker.py
import logging
import socket

from redis.exceptions import ResponseError
from redis_om import NotFoundError

from config import STREAM_REEMBOLSOS, redis
from modelos import Pedido

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("worker-pedidos")

GRUPO = "pedidos"
CONSUMIDOR = socket.gethostname()


def asegurar_grupo() -> None:
    try:
        redis.xgroup_create(STREAM_REEMBOLSOS, GRUPO, id="0", mkstream=True)
    except ResponseError as error:
        if "BUSYGROUP" not in str(error):
            raise


def reembolsar(evento: dict) -> None:
    try:
        pedido = Pedido.get(evento["pedido_id"])
    except NotFoundError:
        log.error("Pedido %s no encontrado", evento["pedido_id"])
        return
    if pedido.estado != "reembolsado":
        pedido.estado = "reembolsado"
        pedido.save()
        log.info("Pedido %s reembolsado", pedido.pk)


def ejecutar() -> None:
    asegurar_grupo()
    while True:
        lotes = redis.xreadgroup(
            GRUPO, CONSUMIDOR, {STREAM_REEMBOLSOS: ">"}, count=20, block=5000
        )
        for _stream, mensajes in lotes or []:
            for mensaje_id, evento in mensajes:
                reembolsar(evento)
                redis.xack(STREAM_REEMBOLSOS, GRUPO, mensaje_id)


if __name__ == "__main__":
    ejecutar()

La comprobación if pedido.estado != "reembolsado" hace que el manejador sea idempotente. Redis Streams garantiza entrega «al menos una vez», no «exactamente una vez»: si el worker procesa un mensaje y muere antes del XACK, ese mensaje volverá a entregarse. Todo consumidor debe tolerar duplicados sin efectos secundarios indeseados.

Este encadenamiento de pasos locales con compensaciones es una versión ligera del patrón saga. En sistemas con muchos pasos se suele introducir un orquestador explícito que conoce el estado de cada saga; para dos servicios, la coreografía basada en eventos que acabas de escribir es suficiente y mucho más simple.

Conectar un frontend en React

Una tienda real necesita una interfaz para listar productos y lanzar pedidos, y React es una de las opciones más habituales. No hace falta construirla entera para entender la integración; lo importante es que el navegador hablará directamente con dos orígenes distintos, por lo que ambos servicios necesitan CORS configurado. Ya lo añadiste en el catálogo; en pedidos se hace igual.

Un componente mínimo que crea un pedido y consulta su estado hasta que deja de estar pendiente podría verse así:

// src/components/Comprar.jsx
import { useState } from "react";

const PEDIDOS_URL = "http://localhost:8001";

export function Comprar({ productoId }) {
    const [estado, setEstado] = useState("listo");

    async function comprar() {
        const resp = await fetch(`${PEDIDOS_URL}/pedidos`, {
            method: "POST",
            headers: { "Content-Type": "application/json" },
            body: JSON.stringify({ producto_id: productoId, cantidad: 1 }),
        });
        const pedido = await resp.json();
        setEstado(pedido.estado);

        const intervalo = setInterval(async () => {
            const r = await fetch(`${PEDIDOS_URL}/pedidos/${pedido.pk}`);
            const actual = await r.json();
            setEstado(actual.estado);
            if (actual.estado !== "pendiente") clearInterval(intervalo);
        }, 1500);
    }

    return <button onClick={comprar}>Comprar ({estado})</button>;
}

El sondeo periódico es la opción más sencilla. Si la interfaz crece, puedes sustituirlo por Server-Sent Events desde FastAPI o por WebSockets, alimentados por el mismo stream de eventos. En producción también es habitual colocar un API gateway o un proxy inverso delante de ambos servicios, de forma que el navegador vea un único origen y CORS deje de ser necesario.

Orquestación local con Docker Compose

Levantar cinco procesos a mano (Redis, dos APIs y dos workers) es tedioso. Docker Compose los describe en un solo archivo. El Dockerfile es idéntico para ambos servicios:

# catalogo/Dockerfile (igual en pedidos/)
FROM python:3.15-slim

ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
RUN useradd --create-home appuser
USER appuser

CMD ["fastapi", "run", "main.py", "--port", "8000"]

El comando fastapi run arranca Uvicorn en modo producción, sin recarga automática y escuchando en 0.0.0.0. El archivo compose.yaml reutiliza la misma imagen para la API y para el worker de cada servicio, cambiando solo el comando:

# compose.yaml
services:
    redis:
        image: redis:8.8
        volumes:
            - redis-datos:/data
        command: ["redis-server", "--appendonly", "yes"]
        healthcheck:
            test: ["CMD", "redis-cli", "PING"]
            interval: 5s
            retries: 5

    catalogo:
        build: ./catalogo
        ports: ["8000:8000"]
        environment:
            REDIS_URL: redis://redis:6379/0
        depends_on:
            redis: { condition: service_healthy }

    catalogo-worker:
        build: ./catalogo
        command: ["python", "worker.py"]
        environment:
            REDIS_URL: redis://redis:6379/0
        depends_on:
            redis: { condition: service_healthy }

    pedidos:
        build: ./pedidos
        ports: ["8001:8000"]
        environment:
            REDIS_URL: redis://redis:6379/0
            CATALOGO_URL: http://catalogo:8000
        depends_on:
            redis: { condition: service_healthy }
            catalogo: { condition: service_started }

    pedidos-worker:
        build: ./pedidos
        command: ["python", "worker.py"]
        environment:
            REDIS_URL: redis://redis:6379/0
        depends_on:
            redis: { condition: service_healthy }

volumes:
    redis-datos:

La versión actual de Compose ya no requiere el campo version al principio del archivo; si lo incluyes, solo obtendrás una advertencia. Arranca todo y escala el worker del catálogo para comprobar que el grupo reparte la carga:

docker compose up --build -d
docker compose up -d --scale catalogo-worker=3
docker compose logs -f catalogo-worker

catalogo-worker-1  | INFO:worker-catalogo:Consumidor 3f2a1c9e escuchando pedidos:completados
catalogo-worker-2  | INFO:worker-catalogo:Consumidor 8b7d4e01 escuchando pedidos:completados
catalogo-worker-3  | INFO:worker-catalogo:Consumidor c49a0f77 escuchando pedidos:completados

Cada réplica usa su hostname como nombre de consumidor, así que Redis las distingue y reparte los mensajes entre ellas sin que debas coordinar nada.

Errores frecuentes y cómo resolverlos

Estos son los problemas que aparecen con más frecuencia al montar este tipo de sistema, junto con su causa habitual.

El primero es un error al crear el grupo de consumidores cuando el worker se reinicia:

redis.exceptions.ResponseError: BUSYGROUP Consumer Group name already exists

Es normal: el grupo se creó en una ejecución anterior. Por eso asegurar_grupo ignora específicamente ese mensaje y relanza cualquier otro error.

El segundo aparece cuando intentas leer de un stream que todavía no existe:

redis.exceptions.ResponseError: NOGROUP No such key 'pedidos:completados' or consumer group 'catalogo'

La solución es crear el grupo con mkstream=True, que genera el stream vacío si hace falta.

Un tercer clásico surge al usar el nombre de host equivocado dentro de contenedores:

httpx.ConnectError: [Errno 111] Connection refused

Dentro de Docker, localhost apunta al propio contenedor, no al host ni a otro servicio. Usa el nombre del servicio definido en Compose (http://catalogo:8000) en la variable CATALOGO_URL.

Si el navegador muestra «blocked by CORS policy» al llamar a la API, revisa que el origen exacto, con esquema y puerto, figure en allow_origins. Un http://localhost:3000 no es lo mismo que http://127.0.0.1:3000.

Cuando el servidor deja de responder durante unos segundos tras cada pedido, casi siempre hay una llamada bloqueante dentro de un async def: time.sleep, requests.get o un cliente de Redis síncrono. Declara el endpoint con def o usa la variante asíncrona de la librería (asyncio.sleep, HTTPX asíncrono, redis.asyncio).

Por último, si los pedidos se quedan en pendiente tras reiniciar el contenedor, la tarea en segundo plano se perdió con el proceso. Revisa la sección siguiente para hacer el flujo resistente a reinicios.

Buenas prácticas para llevarlo a producción

El sistema funciona, pero varias decisiones tomadas por simplicidad merecen revisión antes de exponerlo a tráfico real.

Mueve el procesamiento de pagos a un worker dedicado. En lugar de BackgroundTasks, publica un evento pedidos:creados y deja que un consumidor lo procese. Así el trabajo sobrevive a reinicios, se puede reintentar y escala de forma independiente de la API. Para escenarios con reintentos programados, prioridades o flujos largos, herramientas como Celery, Dramatiq, ARQ o un motor de flujos duraderos como Temporal resultan más adecuadas.

Recupera mensajes huérfanos. Si una réplica muere tras leer un mensaje y antes de confirmarlo, ese mensaje queda en su lista de pendientes. Añade al worker una rutina periódica con XAUTOCLAIM que reclame los mensajes inactivos durante más de cierto tiempo:

siguiente, reclamados, _ = redis.xautoclaim(
    STREAM_COMPLETADOS, GRUPO, CONSUMIDOR, min_idle_time=60_000, start_id="0-0", count=50
)
for mensaje_id, evento in reclamados:
    descontar_existencias(evento)
    redis.xack(STREAM_COMPLETADOS, GRUPO, mensaje_id)

Combina esa rutina con un contador de intentos y una cola de mensajes muertos (un stream separado) para que un mensaje corrupto no se reintente eternamente.

Implementa la salida ordenada de los workers capturando SIGTERM: termina el lote en curso, confirma lo procesado y sal del bucle. Kubernetes y Docker envían esa señal antes de matar el contenedor, y respetarla evita duplicados innecesarios.

Protege la comunicación síncrona. Además del timeout, añade reintentos con retroceso exponencial para errores transitorios y considera un circuit breaker si el catálogo puede degradarse. Nunca reintentes operaciones no idempotentes, como un cobro, sin una clave de idempotencia.

Separa los datos. Compartir una instancia de Redis entre servicios es aceptable en desarrollo, pero en producción cada servicio debería tener su propia base lógica o, mejor, su propia instancia, con usuarios ACL que limiten qué claves y comandos puede usar cada uno. Activa autenticación, TLS y persistencia AOF, y monitoriza la longitud de los streams y el número de mensajes pendientes por grupo con XINFO GROUPS.

Añade observabilidad desde el inicio. Expón un endpoint /salud en cada API, envía logs estructurados en JSON y propaga un identificador de correlación en las cabeceras HTTP y en los eventos. OpenTelemetry tiene instrumentación para FastAPI, HTTPX y redis-py, de modo que puedes ver una compra completa como una sola traza que atraviesa ambos servicios.

Redis Streams frente a otras alternativas de mensajería

Redis Streams no es la única forma de propagar eventos entre servicios. Elegir bien depende del volumen, de las garantías que necesites y de la infraestructura que ya tengas.

En la práctica, la mayoría de los sistemas combinan ambos estilos, igual que en este tutorial: HTTP para consultas que requieren respuesta y eventos para notificar hechos consumados. La regla útil es preguntarse si quien envía necesita saber el resultado para continuar. Si la respuesta es no, un evento reduce el acoplamiento.

En cuanto al framework, Flask sigue siendo válido para servicios síncronos sencillos, Django REST Framework brilla cuando necesitas el ORM y el panel de administración de Django, y Litestar es una alternativa ASGI con una filosofía parecida a FastAPI. FastAPI destaca por su ecosistema, su documentación automática y la integración directa con Pydantic, que hace muy natural compartir esquemas entre la API y los eventos.

Casos de uso donde este enfoque brilla

La combinación de microservicios en Python con FastAPI y Redis Streams encaja especialmente bien en algunos escenarios concretos. Un comercio electrónico pequeño o mediano puede separar catálogo, pedidos, notificaciones y facturación sin introducir un clúster de Kafka. Un backend de aplicaciones móviles puede usar streams para enviar notificaciones push o actualizar cachés de lectura cuando cambian los datos. Las plataformas de IoT aprovechan la baja latencia de Redis para ingerir lecturas de sensores y repartirlas entre consumidores que calculan alertas y agregados.

También es un buen punto de partida para equipos que están extrayendo módulos de un monolito. Puedes publicar eventos desde el monolito existente y construir el nuevo servicio como consumidor, sin modificar las rutas que ya funcionan. Cuando el servicio nuevo madure, el monolito empieza a delegar en él.

Donde no conviene este enfoque es en sistemas que requieren retener eventos durante meses para reprocesarlos, en flujos financieros que necesitan garantías transaccionales fuertes entre servicios o en equipos tan pequeños que el coste operativo de varios servicios supera sus beneficios. En esos casos, un monolito modular bien estructurado suele ser la mejor decisión.

Conclusiones

Has construido un sistema de dos microservicios con FastAPI que se comunican de dos formas complementarias: peticiones HTTP asíncronas con HTTPX cuando el servicio de pedidos necesita un dato al instante, y eventos en Redis Streams cuando basta con notificar que algo ocurrió. Redis 8 cubre en un solo componente la persistencia, gracias a Redis OM y Pydantic 2, y el transporte de eventos, con grupos de consumidores que reparten la carga entre réplicas.

Por el camino actualizaste prácticas que aparecen en muchos tutoriales antiguos: lifespan en lugar de on_event, model_dump en lugar de dict, el comando fastapi run, la imagen oficial de Redis en lugar de Redis Stack y un archivo Compose sin el campo version. También viste por qué el descuento de existencias debe ser atómico, por qué los consumidores tienen que ser idempotentes y cómo una compensación sustituye a una transacción distribuida.

El siguiente paso natural es endurecer el flujo: mover el cobro a un worker propio, añadir XAUTOCLAIM y una cola de mensajes muertos, instrumentar con OpenTelemetry y separar los datos de cada servicio. Con esa base, añadir nuevos servicios, como notificaciones o facturación, se reduce a crear otro grupo de consumidores sobre los eventos que ya publicas.