本文介绍如何通过事件钩子（event hooks）和自定义 transport 实现对 asyncclient 所有 http 调用的请求体与响应体自动日志记录，避免在每个接口中重复编写日志逻辑，提升可维护性与可观测性。
本文介绍如何通过事件钩子（event hooks）和自定义 transport 实现对 asyncclient 所有 http 调用的请求体与响应体自动日志记录，避免在每个接口中重复编写日志逻辑，提升可维护性与可观测性。
HTTPX 的 AsyncClient 本身不直接暴露请求/响应体的完整日志能力，但提供了灵活的扩展机制：事件钩子（event hooks）用于请求阶段，而响应体需借助自定义 AsyncBaseTransport 拦截流式读取并复用内容。下面分步实现一个生产就绪的通用日志方案。
✅ 请求体日志：使用 event_hooks['request']
AsyncClient 支持在请求发出前触发 request 钩子，此时 request.content 已序列化（如 JSON 字符串的 bytes），可直接记录：
import logging
from httpx import AsyncClient, Request
logging.basicConfig(level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s")
async def log_request(request: Request) -> None:
try:
# 尝试解码为 UTF-8 字符串（适用于 JSON/text）
body_str = request.content.decode("utf-8") if request.content else ""
logging.debug(f"[REQUEST] {request.method} {request.url} → body: {body_str[:200]}{'...' if len(body_str) > 200 else ''}")
except UnicodeDecodeError:
logging.debug(f"[REQUEST] {request.method} {request.url} → body: <binary bytes>")
client = AsyncClient(event_hooks={"request": [log_request]})</binary>
⚠️ 注意：request.content 是原始字节，若使用 json= 参数（而非 content=），HTTPX 会自动序列化并设 Content-Type: application/json，此时 request.content 即为合法 JSON bytes；若使用 data=，则为表单编码字节，需按需解析。
✅ 响应体日志：重写 Transport 实现响应体捕获与复用
由于 response.content 是一次性流（await response.aread() 后不可再读），直接在 event_hooks['response'] 中读取会导致后续 .json() 或 .text() 失败。推荐方案是继承 httpx.AsyncHTTPTransport，在 handle_async_request 中拦截响应并缓存内容：
from httpx import AsyncHTTPTransport, Response, Request
import asyncio
class LoggingTransport(AsyncHTTPTransport):
async def handle_async_request(self, request: Request) -> Response:
# 1. 发起原始请求
response = await super().handle_async_request(request)
# 2. 异步读取并缓存响应体（仅一次）
try:
body_bytes = await response.aread()
# 3. 重新构造可复用的响应（保留状态码、headers 等）
new_response = Response(
status_code=response.status_code,
headers=response.headers,
content=body_bytes, # 可多次读取
request=response.request,
extensions=response.extensions,
)
# 4. 记录日志（安全解码，失败则显示 hex 摘要）
try:
body_str = body_bytes.decode("utf-8")
logging.debug(f"[RESPONSE] {response.status_code} ← {response.url} ← body: {body_str[:200]}{'...' if len(body_str) > 200 else ''}")
except UnicodeDecodeError:
logging.debug(f"[RESPONSE] {response.status_code} ← {response.url} ← body: <binary bytes>")
return new_response
except Exception as e:
logging.error(f"Failed to log response body for {response.url}: {e}")
return response
# 使用自定义 Transport 初始化 client
client = AsyncClient(transport=LoggingTransport())</binary>
✅ 完整集成示例（FastAPI 场景）
将上述逻辑整合进你的 FastAPI 应用：
from fastapi import FastAPI, HTTPException
import json
from httpx import AsyncClient, Request, Response
from httpx._transports.default import AsyncHTTPTransport
# ... [LoggingTransport 和 log_request 定义同上] ...
app = FastAPI()
client = AsyncClient(transport=LoggingTransport(), event_hooks={"request": [log_request]})
@app.get("/example_post")
async def example_post():
url = "https://jsonplaceholder.typicode.com/posts"
payload = {"title": "fooxxx", "body": "bar", "userId": 1}
# 自动记录 request.body 和 response.body，无需手动 log
response = await client.post(url, json=payload) # 推荐用 json= 自动序列化 + 设置 header
if response.is_error:
raise HTTPException(status_code=response.status_code, detail="External API error")
return response.json() # 此处 .json() 不会报错 —— body 已缓存
✅ 注意事项与最佳实践
- 性能权衡：响应体缓存会增加内存占用（尤其大文件），生产环境建议限制日志级别（如仅 DEBUG 开启）或添加大小阈值（如 if len(body_bytes)
- 敏感数据脱敏：日志中可能含 token、密码等，务必在 log_request / LoggingTransport 中添加字段过滤逻辑（例如正则匹配 "token": "[^"]+" 并替换）。
- Transport 兼容性：LoggingTransport 继承自 AsyncHTTPTransport，支持代理、SSL 配置等全部原生特性，无功能损失。
-
替代方案对比：
- ❌ 不推荐 response.text / response.json() 后再 log —— 破坏调用方逻辑；
- ❌ 避免全局 monkey patch —— 可维护性差且易冲突；
- ✅ 本方案符合 HTTPX 官方扩展规范，清晰、解耦、可测试。
通过组合事件钩子与自定义 Transport，你获得了完全透明、零侵入、可复用的全链路 HTTP I/O 日志能力 —— 这正是构建可观测微服务的关键基础设施。
