如何扩展并参数化AsyncClient，实现外部调用输入输出体的日志复用？

当然可以实现这个需求，我们可以通过子类化AsyncClient或者利用httpx的事件钩子来封装通用的日志逻辑，让所有使用该客户端的请求自动记录输入和输出，无需在每个接口中重复编写日志代码。

方法一：子类化AsyncClient #

直接继承httpx.AsyncClient，重写核心的send方法，在请求发送前和响应返回后插入日志逻辑：

import json
import logging
from fastapi import FastAPI
from httpx import AsyncClient, Request, Response

# 配置基本日志设置
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

class LoggingAsyncClient(AsyncClient):
    async def send(self, request: Request, **kwargs) -> Response:
        # 记录请求信息
        try:
            request_body = await request.read()
            # 重置请求体，避免后续流程无法读取内容
            request.content = request_body
            logger.info(f"发送请求: URL={request.url}, 方法={request.method}, 请求体={request_body.decode('utf-8') if request_body else '无'}")
        except Exception as e:
            logger.warning(f"读取请求体失败: {str(e)}")
        
        # 执行原始请求逻辑
        response = await super().send(request, **kwargs)
        
        # 记录响应信息
        try:
            response_body = response.text
            logger.info(f"收到响应: URL={request.url}, 状态码={response.status_code}, 响应体={response_body}")
        except Exception as e:
            logger.warning(f"读取响应体失败: {str(e)}")
        
        return response

app = FastAPI()
# 替换为自定义的日志客户端
client = LoggingAsyncClient()

@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参数自动序列化，替代手动dumps
    data = response.json()

    return data

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

方法二：使用httpx事件钩子 #

利用httpx内置的event_hooks参数，在请求发送和响应返回时触发自定义日志函数，无需子类化客户端：

import json
import logging
from fastapi import FastAPI
from httpx import AsyncClient, Request, Response

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

async def log_request(request: Request):
    try:
        body = await request.read()
        request.content = body
        logger.info(f"发送请求: URL={request.url}, 方法={request.method}, 请求体={body.decode('utf-8') if body else '无'}")
    except Exception as e:
        logger.warning(f"读取请求体失败: {str(e)}")

def log_response(response: Response):
    try:
        logger.info(f"收到响应: URL={response.request.url}, 状态码={response.status_code}, 响应体={response.text}")
    except Exception as e:
        logger.warning(f"读取响应体失败: {str(e)}")
    return response

app = FastAPI()
# 创建带日志钩子的客户端
client = AsyncClient(event_hooks={
    "request": [log_request],
    "response": [log_response]
})

@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)
    data = response.json()

    return data

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

关键细节说明 #

• 全局复用：两种方案都能让所有使用该client的请求自动触发日志，无需在每个接口中重复编写日志代码。
• 请求体重置：读取请求体后必须重置request.content，否则httpx后续的请求处理会因为内容已被读取而报错。
• 简化序列化：使用json=payload替代content=json.dumps(payload)，httpx会自动处理JSON序列化和Content-Type头，更简洁可靠。
• 异常防护：添加了读取请求/响应体的异常捕获，避免日志逻辑故障影响正常请求流程。

内容的提问来源于stack exchange，提问作者Levijatanu