使用 AsyncClient 调用外部 API 时，FastApi 记录输入/输出主体

Question

AsyncClient 是否可以扩展和参数化以记录每个外部调用的输入和输出主体，以便在使用相同 AsyncClient 的多个方法中重用？

import json
import logging

from fastapi import FastAPI, HTTPException
from httpx import AsyncClient

# Set up basic configuration for logging
logging.basicConfig(level=logging.DEBUG)

app = FastAPI()
client = AsyncClient()


@app.get("/example_post")
async def example_post():
    url = "https://jsonplaceholder.typicode.com/posts"  # Free fake and reliable API for testing and prototyping.
    payload = {
        "title": 'fooxxx',
        "body": 'bar',
        "userId": 1,
    }

    # Perform the HTTP POST request; would like to log input payload and output data response
    response = await client.post(url, content=json.dumps(payload))

    # Parse response as JSON
    data = response.json()

    # Searching for universal solution that can log request and response of every api call instead of logging manually
    logging.info(f"Received response: {data}")

    return data


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)

Answer 1

请求正文

在记录请求正文时，您应该考虑事件挂钩。

这是文档：https://www.python-httpx.org/advanced/event-hooks/。

以下是实现它的方法：

from httpx import AsyncClient
from httpx import Response, Request

async def log_request(request: Request):
    print(f"Request event hook: {request.method} {request.url} {request.content} - Waiting for response")

client = AsyncClient(event_hooks={'request': [log_request]})

响应正文

对于响应主体，事情有点复杂，因为主体是一个只能被消耗一次的流。

为了解决这个问题，您可以考虑编写自己的传输层。你可以参考这个答案：https://github.com/encode/httpx/discussions/3073#discussioncomment-8263619