Can AsyncClient be extended and parametrized to log the body of input and output of every external call for reuse in multiple methods that use the same 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)
When it comes to logging the request body, you should consider event hooks.
Here's the doc : https://www.python-httpx.org/advanced/event-hooks/.
And here's how you could implement it :
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]})
As for the response body, things are a little more complicated, as the body is a stream that can only be consumed once.
To get around this, you could consider writing your own transport layer. You can refer to this answer: https://github.com/encode/httpx/discussions/3073#discussioncomment-8263619