FastAPI & Python 3.12: Building Async Microservices with Pydantic v2

2. Step by step: call another microservice with httpx and a lifespan client

A single FastAPI service is a demo; microservices begin at the first outbound call. The production pattern has three rules: one shared async HTTP client for the whole app, created and closed with the application lifespan; a timeout on every call; and Pydantic models on both sides of the wire. Here is the pattern, using httpx — the async-native HTTP library that matches FastAPI’s ASGI model.

Step 1: create the shared client in the 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

Creating an AsyncClient per request — the mistake every first draft makes — throws away connection pooling and adds a TLS handshake to every call. The lifespan client is created once, injected as a dependency, and closed cleanly on shutdown.

Step 2: validate, call, respond — all typed

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. A complete worked example

Run the pair locally and watch the contract hold. Start the inventory stub and the order service, then place an order:

# 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}'

Three outcomes demonstrate the design. A valid order returns 201 with the OrderOut body — nothing extra leaks, because the response model filters the payload (the discipline the deep-dive insists on). An order for more than the available stock returns 409 with a readable detail, decided by this service from the peer’s data. And if you kill the inventory service and retry, the order service answers 502 within the client’s five-second budget instead of hanging — the failure is contained at the boundary where it belongs, and /docs shows all three responses because FastAPI generates them from the same models that enforce them.

Step 3: test the boundary without the network

The dependency-injection design pays off in tests: FastAPI’s dependency_overrides lets you swap the real inventory client for one backed by httpx.MockTransport, so the 201, 409, and error paths are testable in milliseconds with no services running:

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

Note what the second test proves: validation happens at this service’s edge, before a single byte reaches the peer. That ordering — validate, then call, then map errors — is the whole microservice contract in miniature.

Step 4: freeze the contract for consumers

The last habit costs one command. Because the Pydantic models are the contract, export the generated OpenAPI schema and commit it next to the code. Consumer teams generate their clients from this file, and your CI can diff it on every pull request — a breaking change to a response model then fails the build loudly instead of surprising a peer service in production:

# 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

That diff is the cheapest contract test ever written, and it closes the loop this tutorial opened with: the models validate at runtime, document at /docs, and now also guard themselves in version control.

4. Troubleshooting

• "RuntimeError: This event loop is already running" or a coroutine was never awaited. Somewhere an async call lost its await — classically, calling the httpx client but returning the coroutine object, which Pydantic then fails to serialize with a confusing error. Search the traceback for the endpoint, and make every client call await resp = await client.get(...).
• ConnectError on the first request after deployment. The peer service is not up yet, or the base_url host only resolves inside your compose/Kubernetes network. Verify name resolution from inside the order service’s environment first; then add startup resilience — a retry with backoff on ConnectError — rather than longer timeouts.
• 422 errors on payloads that look correct. Print the actual body FastAPI received before blaming the model: the usual culprits are a JSON string sent instead of an object (double-encoded by the caller), camelCase keys against snake_case fields, or an integer arriving as a string from a form post. The /docs "Try it out" payload is the ground truth for what the model expects.
• Latency grows linearly with traffic. Check for per-request AsyncClient creation (step 1 exists to prevent it) and for a synchronous call hiding in the path — one requests.get or time.sleep inside an async def stalls the event loop for every concurrent request, exactly the failure this tutorial opened with.