import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response获取请求和响应
@app.middleware("http")
async def log_request_and_response(request: Request, call_next) -> Response:
data = {
"client": request.client[0],
"method": request.method,
"path": request.url.path,
"params": dict(request.query_params),
"body": await request.body(),
}
response = await call_next(request)
response_body = b""
async for chunk in response.body_iterator:
response_body += chunk
data["data"] = response_body.decode()[:256]
ilog.info(data)
return Response(
content=response_body,
status_code=response.status_code,
headers=dict(response.headers),
media_type=response.media_type,
)
部分 POST 请求无响应的解决方案：
async def set_body(request: Request, body: bytes):
async def receive() -> Message:
return {"type": "http.request", "body": body}
request._receive = receive
async def get_body(request: Request) -> bytes:
body = await request.body()
await set_body(request, body)
return body
@app.middleware("http")
async def app_entry(request: Request, call_next):
await set_body(request, request.body())
print(await get_body(request))
response = await call_next(request)
return response