本文介绍如何通过事件钩子（event hooks）和自定义传输层，为 httpx.AsyncClient 实现全自动、可复用的请求体（input）与响应体（output）日志记录，避免在每个 API 调用处手动打印，提升调试效率与代码一致性。
本文介绍如何通过事件钩子（event hooks）和自定义传输层，为 `httpx.asyncclient` 实现全自动、可复用的请求体（input）与响应体（output）日志记录，避免在每个 api 调用处手动打印，提升调试效率与代码一致性。
在微服务或网关类应用中，对所有外部 HTTP 调用进行结构化日志记录（尤其是请求/响应原始 body）是可观测性的关键实践。httpx.AsyncClient 本身不内置 body 日志能力，但提供了灵活的扩展机制：事件钩子（event hooks）用于请求侧，而响应体需结合自定义 AsyncBaseTransport 实现流重放——因为响应体是单次消费的异步流（response.aiter_bytes()），直接 .read() 会导致后续解析失败。
✅ 请求体日志：使用 event_hooks['request']
AsyncClient 支持在请求发出前触发回调函数，此时 request.content 已序列化为字节（如 JSON 字符串的 UTF-8 编码），可安全读取并记录：
import logging
import json
from httpx import AsyncClient, Request
logging.basicConfig(level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s")
async def log_request(request: Request):
# 尝试安全解码并格式化请求体（仅适用于 text-based content）
try:
body = request.content.decode("utf-8")
if "application/json" in request.headers.get("content-type", ""):
body = json.dumps(json.loads(body), indent=2, ensure_ascii=False)
logging.debug(f"[REQUEST] {request.method} {request.url}\n{body}")
except (UnicodeDecodeError, json.JSONDecodeError, ValueError):
logging.debug(f"[REQUEST] {request.method} {request.url} (binary content, {len(request.content)} bytes)")
client = AsyncClient(
event_hooks={"request": [log_request]}
)⚠️ 注意：request.content 是 bytes 类型，可能包含二进制数据（如文件上传）。上述示例做了容错处理，仅对 JSON 类型尝试美化输出。
✅ 响应体日志：自定义 Transport 拦截并缓存流
由于响应体是异步迭代器，必须在首次消费前“拦截并复制”一份用于日志。标准 AsyncClient 不支持此操作，因此需继承 httpx.AsyncBaseTransport 并重写 handle_async_request 方法：
import asyncio
from httpx import AsyncBaseTransport, Request, Response, AsyncClient
from httpx._transports.default import AsyncHTTPTransport
class LoggingTransport(AsyncBaseTransport):
def __init__(self, transport: AsyncBaseTransport):
self._transport = transport
async def handle_async_request(self, request: Request) -> Response:
# 1. 先执行原始请求
response = await self._transport.handle_async_request(request)
# 2. 安全读取响应体（仅限小到中等 size；生产环境建议加 size 限制）
try:
# 使用 .aread() 获取完整 body（需确保 transport 支持）
body = await response.aread()
# 3. 重新构造可重复使用的响应（关键：将 body 注入 response.stream）
from httpx._models import ByteStream
response._content = body # 缓存原始内容
response.stream = ByteStream(body) # 重置 stream 为可重放流
# 4. 记录日志
try:
decoded = body.decode("utf-8")
if "application/json" in response.headers.get("content-type", ""):
decoded = json.dumps(json.loads(decoded), indent=2, ensure_ascii=False)
logging.debug(f"[RESPONSE] {response.status_code} {response.url}\n{decoded}")
except (UnicodeDecodeError, json.JSONDecodeError):
logging.debug(f"[RESPONSE] {response.status_code} {response.url} ({len(body)} bytes)")
except Exception as e:
logging.warning(f"Failed to log response body: {e}")
return response
# 创建带日志功能的 client
base_transport = AsyncHTTPTransport()
client = AsyncClient(transport=LoggingTransport(base_transport))✅ 完整集成示例（FastAPI 中复用）
将上述逻辑封装后，即可在任意路由中无感使用：
from fastapi import FastAPI, HTTPException
import json
from httpx import AsyncClient
app = FastAPI()
# ✅ 全局复用的带日志 client
client = AsyncClient(
transport=LoggingTransport(AsyncHTTPTransport()),
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}
response = await client.post(url, json=payload) # 注意：使用 json= 自动设 header & serialize
return response.json()? 关键注意事项
- 性能权衡：响应体日志会额外读取并缓存整个 body，对大文件或高吞吐场景不推荐；生产环境建议添加 max_body_size=1024*1024（1MB）限制，并跳过超限响应。
- Content-Type 判断：日志美化仅适用于 application/json、text/* 等文本类型，二进制响应（如图片、PDF）应仅记录长度。
- 线程/协程安全：LoggingTransport 是协程安全的，但若多处共享同一 client 实例，日志本身天然并发安全（logging 模块已处理）。
- 替代方案：如只需调试，可临时启用 httpx 内置日志（logging.getLogger("httpx").setLevel(logging.DEBUG)），但它不显示原始 body，仅含 headers 和状态码。
通过组合事件钩子与自定义 Transport，你获得了一个轻量、可复用、符合 httpx 设计哲学的全自动日志方案——从此告别每个 await client.post() 后的手动 logging.info(...)。
