Microservicios con FastAPI y Python: APIs Asíncronas de Alto Rendimiento

2. Paso a paso: llama a otro microservicio con httpx y un cliente en el lifespan

Un único servicio FastAPI es una demo; los microservicios empiezan en la primera llamada saliente. El patrón de producción se resume en tres reglas: un solo cliente HTTP asíncrono compartido por toda la aplicación, creado y cerrado junto con el lifespan; un tiempo de espera en cada llamada; y modelos Pydantic a ambos lados de la conexión. Aquí lo tienes con httpx, la biblioteca HTTP pensada para async que encaja con el modelo ASGI de FastAPI.

Paso 1: crea el cliente compartido en el lifespan

from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI, Request


@asynccontextmanager
async def lifespan(app: FastAPI):
    # One client, one connection pool, for the life of the process.
    app.state.inventory = httpx.AsyncClient(
        base_url="http://inventory-service:8000",
        timeout=5.0,
    )
    yield
    await app.state.inventory.aclose()


app = FastAPI(title="Order Service", lifespan=lifespan)


async def get_inventory(request: Request) -> httpx.AsyncClient:
    return request.app.state.inventory

Crear un AsyncClient por cada petición —el error que comete todo primer borrador— desperdicia el pool de conexiones y suma un handshake TLS a cada llamada. El cliente del lifespan se crea una sola vez, se inyecta como dependencia y se cierra de forma limpia al apagar la aplicación.

Paso 2: valida, llama y responde, todo con tipos

from fastapi import Depends, HTTPException, status
from pydantic import BaseModel, Field


class OrderCreate(BaseModel):
    sku: str = Field(..., min_length=1)
    quantity: int = Field(..., gt=0)


class OrderOut(BaseModel):
    sku: str
    quantity: int
    status: str


@app.post("/orders", response_model=OrderOut,
          status_code=status.HTTP_201_CREATED)
async def create_order(
    payload: OrderCreate,
    client: httpx.AsyncClient = Depends(get_inventory),
):
    try:
        resp = await client.get(f"/stock/{payload.sku}")
        resp.raise_for_status()
    except httpx.HTTPError:
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail="Inventory service unavailable",
        )

    if resp.json()["available"] < payload.quantity:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Insufficient stock",
        )

    return OrderOut(sku=payload.sku, quantity=payload.quantity,
                    status="reserved")

3. Un ejemplo completo

Ejecuta la pareja de servicios en local y observa cómo se mantiene el contrato. Arranca el stub de inventario y el servicio de órdenes, y después haz un pedido:

# Terminal 1 - the inventory service (a peer FastAPI app)
uvicorn inventory_service:app --port 8001

# Terminal 2 - the order service from this tutorial
uvicorn order_service:app --port 8000

# Terminal 3 - place an order
curl -s -X POST http://localhost:8000/orders \
  -H "Content-Type: application/json" \
  -d '{"sku": "WIDGET-1", "quantity": 2}'

Hay tres resultados que demuestran el diseño. Un pedido válido devuelve 201 con el cuerpo OrderOut, sin que se filtre nada extra, porque el modelo de respuesta filtra la carga útil —la disciplina en la que insiste la sección en profundidad—. Un pedido que supera el stock disponible devuelve 409 con un detalle legible, decidido por este servicio a partir de los datos del otro. Y si detienes el servicio de inventario y reintentas, el servicio de órdenes responde 502 dentro del margen de cinco segundos del cliente en lugar de quedarse colgado: el fallo queda contenido en la frontera, que es donde debe estar, y /docs muestra las tres respuestas porque FastAPI las genera desde los mismos modelos que las hacen cumplir.

Paso 3: prueba la frontera sin tocar la red

El diseño con inyección de dependencias rinde frutos en las pruebas: dependency_overrides de FastAPI te permite sustituir el cliente real de inventario por uno respaldado por httpx.MockTransport, de modo que los caminos 201, 409 y de error se prueban en milisegundos y sin ningún servicio en marcha:

import httpx
from fastapi.testclient import TestClient

from order_service import app, get_inventory


async def fake_inventory_handler(request: httpx.Request) -> httpx.Response:
    # The peer service, simulated: 10 widgets always in stock
    return httpx.Response(200, json={"available": 10})


def override_inventory() -> httpx.AsyncClient:
    transport = httpx.MockTransport(fake_inventory_handler)
    return httpx.AsyncClient(transport=transport, base_url="http://inventory")


app.dependency_overrides[get_inventory] = override_inventory
client = TestClient(app)


def test_order_reserved_when_stock_covers_it():
    resp = client.post("/orders", json={"sku": "WIDGET-1", "quantity": 2})
    assert resp.status_code == 201
    assert resp.json()["status"] == "reserved"


def test_order_rejected_when_quantity_is_invalid():
    resp = client.post("/orders", json={"sku": "WIDGET-1", "quantity": 0})
    assert resp.status_code == 422   # Pydantic rejects it before any call

Fíjate en lo que demuestra la segunda prueba: la validación ocurre en el borde de este servicio, antes de que un solo byte llegue al otro extremo. Ese orden —validar, después llamar y por último mapear los errores— es todo el contrato de microservicios en miniatura.

Paso 4: congela el contrato para quien lo consume

El último hábito cuesta un solo comando. Como los modelos Pydantic son el contrato, exporta el esquema OpenAPI generado y guárdalo junto al código. Los equipos consumidores generan sus clientes desde ese archivo, y tu CI puede compararlo en cada pull request: un cambio incompatible en un modelo de respuesta hará fallar la compilación con ruido, en lugar de sorprender a otro servicio ya en producción:

# Snapshot the contract the models currently generate
curl -s http://localhost:8000/openapi.json -o openapi.json

# In CI: fail the build if the contract changed without a version bump
git diff --exit-code openapi.json

Esa comparación es la prueba de contrato más barata que se ha escrito, y cierra el círculo con el que abrió este tutorial: los modelos validan en tiempo de ejecución, documentan en /docs y ahora también se protegen a sí mismos en el control de versiones.

4. Solución de errores

• «RuntimeError: This event loop is already running» o una corrutina que nunca fue esperada. En algún punto una llamada asíncrona perdió su await; el caso clásico es llamar al cliente httpx pero devolver el objeto corrutina, que Pydantic después no logra serializar y devuelve un error confuso. Busca el endpoint en el traceback y asegúrate de que cada llamada al cliente sea await resp = await client.get(...).
• ConnectError en la primera petición tras desplegar. El servicio par todavía no arrancó, o el host de base_url solo se resuelve dentro de tu red de Compose o Kubernetes. Verifica primero la resolución de nombres desde el propio entorno del servicio de órdenes; después añade resiliencia al arranque —reintentos con backoff ante ConnectError— en lugar de alargar los tiempos de espera.
• Errores 422 con cargas que parecen correctas. Imprime el cuerpo real que recibió FastAPI antes de culpar al modelo: los culpables habituales son una cadena JSON enviada en vez de un objeto —doblemente codificada por quien llama—, claves en camelCase frente a campos en snake_case, o un entero que llega como cadena desde un formulario. La carga de «Try it out» en /docs es la referencia fiel de lo que el modelo espera.
• La latencia crece en proporción al tráfico. Revisa si se está creando un AsyncClient por petición —el paso 1 existe justo para evitarlo— y si hay alguna llamada síncrona escondida en el camino: un solo requests.get o un time.sleep dentro de un async def detiene el bucle de eventos para todas las peticiones concurrentes, exactamente el fallo con el que abrió este tutorial.