middleware

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
    )

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