FASTAPI custom middleware getting body of request inside

We Are Going To Discuss About FASTAPI custom middleware getting body of request inside. So lets Start this Python Article.

FASTAPI custom middleware getting body of request inside

  1. How to solve FASTAPI custom middleware getting body of request inside

    You need to await on the request so the request object is infact ready to be read. This is how I achieved it.
    class RequestContextLogMiddleware(BaseHTTPMiddleware): async def set_body(self, request: Request): receive_ = await request._receive() async def receive() -> Message: return receive_ request._receive = receive async def dispatch(self, request: Request, call_next: RequestResponseEndpoint): await self.set_body(request) body = await request.body() jsonbody = await request.json() id_ = jsonbody['external_id'] response = await call_next(request) return response

  2. FASTAPI custom middleware getting body of request inside

    You need to await on the request so the request object is infact ready to be read. This is how I achieved it.
    class RequestContextLogMiddleware(BaseHTTPMiddleware): async def set_body(self, request: Request): receive_ = await request._receive() async def receive() -> Message: return receive_ request._receive = receive async def dispatch(self, request: Request, call_next: RequestResponseEndpoint): await self.set_body(request) body = await request.body() jsonbody = await request.json() id_ = jsonbody['external_id'] response = await call_next(request) return response

Solution 1

You need to await on the request so the request object is infact ready to be read. This is how I achieved it.

class RequestContextLogMiddleware(BaseHTTPMiddleware):

    async def set_body(self, request: Request):
        receive_ = await request._receive()

        async def receive() -> Message:
            return receive_

        request._receive = receive

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint):
        await self.set_body(request)
        body = await request.body()
        jsonbody = await request.json()
        id_ = jsonbody['external_id']
        response = await call_next(request)    
        return response

Original Author Saif Asif Of This Content

Solution 2

Solution of getting request body references FastAPI – Custom APIRoute class in a router

class CustomRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            body = await request.body()
            response: Response = await original_route_handler(request)
            return response

        return custom_route_handler

Original Author Jedore Of This Content

Solution 3

I would recommend using a router instead. Reference this github issue

Here is an example

app = FastAPI()
api_router = APIRouter()


async def log_request_info(request: Request):
    request_body = await request.json()

    logger.info(
        f"{request.method} request to {request.url} metadata\n"
        f"\tHeaders: {request.headers}\n"
        f"\tBody: {request_body}\n"
        f"\tPath Params: {request.path_params}\n"
        f"\tQuery Params: {request.query_params}\n"
        f"\tCookies: {request.cookies}\n"
    )


@api_router.get("/", summary="Status")
async def status_get():
    logger.debug('Status requested')
    return {'status': 'OK'}


@api_router.post("/", )
def status_post(urls: Optional[List[str]] = None):
    logger.debug('Status requested')
    return {'status': 'OK'}


app.include_router(api_router, dependencies=[Depends(log_request_info)])

Original Author Will Udstrand Of This Content

Solution 4

Quoting FastAPI Doc about “Details about the Request object“:

As FastAPI is actually Starlette underneath, with a layer of several tools on top, you can use Starlette’s Request object directly when you need to.

And the starlette doc about the request body object says:

There are a few different interfaces for returning the body of the request:

  • The request body as bytes: await request.body()
  • The request body, parsed as form data or multipart: await request.form()
  • The request body, parsed as JSON: await request.json()
@app.middleware("http")
async def TestCustomMiddleware(request: Request, call_next):
    the_headers = request.headers
    the_body = await request.json()

    print(the_headers)
    print(the_body)

    response = await call_next(request)

    return response

Tested with curl:

curl -X POST \
  -H "Content-Type: application/json" \
  -d '{"test": "Also works!"}' \
    http://localhost:8000/foo

Original Author prietosanti Of This Content

Conclusion

So This is all About This Tutorial. Hope This Tutorial Helped You. Thank You.

Also Read,

ittutorial team

I am an Information Technology Engineer. I have Completed my MCA And I have 4 Year Plus Experience, I am a web developer with knowledge of multiple back-end platforms Like PHP, Node.js, Python and frontend JavaScript frameworks Like Angular, React, and Vue.

Leave a Comment