Chapter 14 · Middleware and Event Hooks

Chapter goals: Master httpx's middleware mechanism and event hooks, implementing common features like logging, authentication injection, and response handling.

14.1 Event Hooks: Request and Response Interception Points

httpx provides an event_hooks mechanism that allows you to insert custom logic before requests are sent and after responses are received. This is the simplest way to implement cross-client features like logging, authentication, and monitoring.

import httpx
from datetime import datetime

# Request hook: inject timestamp before sending
def add_timestamp(request: httpx.Request) -> None:
    request.headers["X-Request-Timestamp"] = datetime.utcnow().isoformat()

# Response hook: log request results
def log_response(response: httpx.Response) -> None:
    req = response.request
    print(f"[{req.method} {req.url}] → {response.status_code}")

# Assemble client
client = httpx.Client(
    event_hooks={
        "request": [add_timestamp],
        "response": [log_response],
    }
)

with client:
    resp = client.get("https://httpbin.org/get")
    print(f"Response body first 50 chars: {resp.text[:50]}")

Async Note

When using httpx.AsyncClient, event hooks must be async functions:

async def async_log_response(response: httpx.Response) -> None:
    await some_async_operation()
    print(f"Status: {response.status_code}")

async with httpx.AsyncClient(event_hooks={"response": [async_log_response]}) as client:
    ...

14.2 Response Hooks: Force Error Handling

Event hooks can not only observe but also modify response behavior. The most common use case is forcing all responses through raise_for_status():

import httpx

def enforce_status(response: httpx.Response) -> None:
    """Automatically throw exception for any 4xx/5xx response"""
    response.raise_for_status()

# Enable globally: no need to manually check status code each time
client = httpx.Client(
    event_hooks={"response": [enforce_status]}
)

# The following code will automatically throw httpx.HTTPStatusError on 404
try:
    resp = client.get("https://httpbin.org/status/404")
except httpx.HTTPStatusError as e:
    print(f"Request failed: {e.response.status_code}")

Note Response Body Read Timing

Response hooks execute before the response body is read. If the hook needs to access the response body, you must explicitly call response.read() or async response.aread().

14.3 Custom Transport: Underlying Middleware Implementation

httpx's "middleware" concept is implemented through custom transport (Custom Transport). The transport layer is responsible for actual network I/O. Replacing it allows intercepting and modifying all requests/responses.

import httpx

class LoggingTransport(httpx.BaseTransport):
    """Transport middleware that logs all requests and responses"""
    
    def __init__(self, transport: httpx.BaseTransport):
        self._transport = transport
    
    def handle_request(self, request: httpx.Request) -> httpx.Response:
        print(f"→ {request.method} {request.url}")
        response = self._transport.handle_request(request)
        print(f"← {response.status_code}")
        return response
    
    def close(self) -> None:
        self._transport.close()
    
    def handle_async_request(self, request: httpx.Request) -> httpx.Response:
        raise NotImplementedError("Use AsyncLoggingTransport")


# Attach to client
transport = httpx.HTTPTransport()
logging_transport = LoggingTransport(transport)
client = httpx.Client(transport=logging_transport)

For async scenarios, inherit httpx.AsyncBaseTransport:

class AsyncLoggingTransport(httpx.AsyncBaseTransport):
    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
        print(f"→ ASYNC {request.method} {request.url}")
        response = await self._transport.handle_async_request(request)
        print(f"← {response.status_code}")
        return response

14.4 Authentication Middleware Example

A practical authentication middleware pattern - automatically adding Bearer Token for specific domains:

import httpx

class AuthMiddleware(httpx.BaseTransport):
    def __init__(self, transport: httpx.BaseTransport, api_key: str):
        self._transport = transport
        self._api_key = api_key
    
    def handle_request(self, request: httpx.Request) -> httpx.Response:
        # Only add auth header for specific domains
        if "api.example.com" in str(request.url):
            request.headers["Authorization"] = f"Bearer {self._api_key}"
        return self._transport.handle_request(request)
    
    def close(self) -> None:
        self._transport.close()


# Usage
base_transport = httpx.HTTPTransport()
auth_transport = AuthMiddleware(base_transport, api_key="sk-xxx")
client = httpx.Client(transport=auth_transport)

# All requests to api.example.com automatically carry auth header
resp = client.get("https://api.example.com/users")

14.5 Using mounts to Route Different Domains

httpx supports routing to different transports by domain, suitable for mixing clients with different configurations:

import httpx

# Normal requests: no special configuration
default_transport = httpx.HTTPTransport()

# API requests: with authentication and retry
api_transport = httpx.HTTPTransport(
    retries=3,
    auth=("user", "pass")
)

# Internal services: direct Unix Socket connection
uds_transport = httpx.HTTPTransport(uds="/var/run/api.sock")

client = httpx.Client(
    mounts={
        "all://": default_transport,           # Default transport
        "all://api.example.com": api_transport,  # API domain uses auth transport
        "all://+internal.local": uds_transport,  # Internal services use UDS
    }
)

14.6 Chapter Summary

• Event hooks (event_hooks) are the simplest way to insert request/response logic;
• Hook functions can modify request/response objects to implement injection, validation, logging;
• Async clients must use async hook functions;
• Custom transport (BaseTransport) is the underlying way to implement complete middleware;
• mounts parameter supports routing to different transport configurations by domain.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. In httpx event hooks, when is the response hook called?

2. What type must event hook functions be for AsyncClient?

3. Which base class must httpx custom transport inherit?

4. What does mounts parameter {"all://api.example.com": transport} do?

🛠️ Hands-on Practice

1. Implement a request delay injection middleware: add X-Request-ID (UUID) to request headers, and print the ID in response hooks.
2. Write a RetryOn5xx transport middleware: automatically retry 5xx responses up to 3 times with 1-second intervals.
3. Configure client with mounts: api.example.com uses authenticated transport, other domains use default transport.

After completing the exercises, proceed to Next chapter: pytest Integration Testing.