中间件

middleware

2023-11-24

1
import time
2
3
from fastapi import FastAPI, Request
4
5
app = FastAPI()
6
7
8
@app.middleware("http")
9
async def add_process_time_header(request: Request, call_next):
10
start_time = time.time()
11
response = await call_next(request)
12
process_time = time.time() - start_time
13
response.headers["X-Process-Time"] = str(process_time)
14
return response

获取请求和响应

1
@app.middleware("http")
2
async def log_request_and_response(request: Request, call_next) -> Response:
3
data = {
4
"client": request.client[0],
5
"method": request.method,
6
"path": request.url.path,
7
"params": dict(request.query_params),
8
"body": await request.body(),
9
}
10
11
response = await call_next(request)
12
response_body = b""
13
async for chunk in response.body_iterator:
14
response_body += chunk
15
16
data["data"] = response_body.decode()[:256]
17
ilog.info(data)
18
19
return Response(
20
content=response_body,
21
status_code=response.status_code,
22
headers=dict(response.headers),
23
media_type=response.media_type,
24
)

部分 POST 请求无响应的解决方案:

1
async def set_body(request: Request, body: bytes):
2
async def receive() -> Message:
3
return {"type": "http.request", "body": body}
4
request._receive = receive
5
6
async def get_body(request: Request) -> bytes:
7
body = await request.body()
8
await set_body(request, body)
9
return body
10
11
@app.middleware("http")
12
async def app_entry(request: Request, call_next):
13
14
await set_body(request, request.body())
15
16
print(await get_body(request))
17
18
response = await call_next(request)
19
return response