第 14 章 · 中间件与事件钩子

本章目标：掌握 httpx 的中间件机制和事件钩子，实现日志记录、认证注入、响应处理等通用功能。

14.1 事件钩子：请求与响应的拦截点

httpx 提供 event_hooks 机制，允许你在请求发送前和响应返回后插入自定义逻辑。这是实现跨客户端功能（如日志、认证、监控）最简洁的方式。

import httpx
from datetime import datetime

# 请求钩子：在发送前注入时间戳
def add_timestamp(request: httpx.Request) -> None:
    request.headers["X-Request-Timestamp"] = datetime.utcnow().isoformat()

# 响应钩子：记录请求结果
def log_response(response: httpx.Response) -> None:
    req = response.request
    print(f"[{req.method} {req.url}] → {response.status_code}")

# 组装客户端
client = httpx.Client(
    event_hooks={
        "request": [add_timestamp],
        "response": [log_response],
    }
)

with client:
    resp = client.get("https://httpbin.org/get")
    print(f"响应体前 50 字符: {resp.text[:50]}")

异步注意

使用 httpx.AsyncClient 时，事件钩子必须是异步函数：

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 响应钩子：强制错误处理

事件钩子不仅可以观察，还能修改响应行为。最常见的用例是强制所有响应都通过 raise_for_status()：

import httpx

def enforce_status(response: httpx.Response) -> None:
    """任何 4xx/5xx 响应自动抛出异常"""
    response.raise_for_status()

# 全局启用：不再需要每次手动检查状态码
client = httpx.Client(
    event_hooks={"response": [enforce_status]}
)

# 以下代码遇到 404 会自动抛出 httpx.HTTPStatusError
try:
    resp = client.get("https://httpbin.org/status/404")
except httpx.HTTPStatusError as e:
    print(f"请求失败: {e.response.status_code}")

注意响应体读取时机

响应钩子在响应体读取之前执行。若钩子内需要访问响应体，必须显式调用 response.read() 或异步的 response.aread()。

14.3 自定义传输：中间件的底层实现

httpx 的"中间件"概念通过**自定义传输（Custom Transport）**实现。传输层负责实际的网络 I/O，替换它可以拦截和修改所有请求/响应。

import httpx

class LoggingTransport(httpx.BaseTransport):
    """记录所有请求和响应的传输中间件"""
    
    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("请使用 AsyncLoggingTransport")


# 挂载到客户端
transport = httpx.HTTPTransport()
logging_transport = LoggingTransport(transport)
client = httpx.Client(transport=logging_transport)

对于异步场景，需要继承 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 认证中间件示例

一个实用的认证中间件模式——自动为特定域名添加 Bearer Token：

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:
        # 只为特定域名添加认证头
        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()


# 使用方式
base_transport = httpx.HTTPTransport()
auth_transport = AuthMiddleware(base_transport, api_key="sk-xxx")
client = httpx.Client(transport=auth_transport)

# 所有对 api.example.com 的请求自动携带认证头
resp = client.get("https://api.example.com/users")

14.5 使用 mounts 路由不同域名

httpx 支持按域名路由到不同的传输，适合混合使用不同配置的客户端：

import httpx

# 普通请求：无特殊配置
default_transport = httpx.HTTPTransport()

# API 请求：带认证和重试
api_transport = httpx.HTTPTransport(
    retries=3,
    auth=("user", "pass")
)

# 内部服务：直连 Unix Socket
uds_transport = httpx.HTTPTransport(uds="/var/run/api.sock")

client = httpx.Client(
    mounts={
        "all://": default_transport,           # 默认传输
        "all://api.example.com": api_transport,  # API 域名用认证传输
        "all://+internal.local": uds_transport,  # 内部服务用 UDS
    }
)

14.6 本章小结

• 事件钩子（event_hooks）是插入请求/响应逻辑的最简方式；
• 钩子函数可修改 request/response 对象，实现注入、验证、日志；
• 异步客户端必须使用异步钩子函数；
• 自定义传输（BaseTransport）是实现完整中间件的底层方式；
• mounts 参数支持按域名路由到不同传输配置。

🧪 随堂测验

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

1. httpx 事件钩子中，响应钩子在哪个时机被调用？

2. AsyncClient 的事件钩子函数必须是什么类型？

3. httpx 自定义传输需要继承哪个基类？

4. mounts 参数 {"all://api.example.com": transport} 的作用是？

🛠️ 动手实践

1. 实现一个请求延迟注入中间件：在请求头中添加 X-Request-ID（UUID），并在响应钩子中打印该 ID。
2. 编写一个 RetryOn5xx 传输中间件：对 5xx 响应自动重试最多 3 次，每次间隔 1 秒。
3. 用 mounts 配置客户端：api.example.com 使用带认证的传输，其他域名使用默认传输。

完成练习后，进入下一章：pytest 集成测试。