import gzip

from typing import Callable, List

from fastapi import Body, FastAPI

from fastapi.routing import APIRoute

from starlette.requests import Request

from starlette.responses import Response

class GzipRequest(Request):

async def body(self) -> bytes:

if not hasattr(self, "_body"):

body = await super().body()

if "gzip" in self.headers.getlist("Content-Encoding"):

body = gzip.decompress(body)

self._body = body

return self._body

class GzipRoute(APIRoute):

def get_route_handler(self) -> Callable:

original_route_handler = super().get_route_handler()

async def custom_route_handler(request: Request) -> Response:

request = GzipRequest(request.scope, request.receive)

return await original_route_handler(request)

return custom_route_handler

app = FastAPI()

app.router.route_class = GzipRoute

async def sum_numbers(numbers: List[int] = Body(...)):

return {"sum": sum(numbers)}

from typing import Callable, List

from fastapi import Body, FastAPI, HTTPException

from fastapi.exceptions import RequestValidationError

from fastapi.routing import APIRoute

from starlette.requests import Request

from starlette.responses import Response

class ValidationErrorLoggingRoute(APIRoute):

def get_route_handler(self) -> Callable:

original_route_handler = super().get_route_handler()

async def custom_route_handler(request: Request) -> Response:

try:

return await original_route_handler(request)

except RequestValidationError as exc:

body = await request.body()

detail = {"errors": exc.errors(), "body": body.decode()}

raise HTTPException(status_code=422, detail=detail)

return custom_route_handler

app = FastAPI()

app.router.route_class = ValidationErrorLoggingRoute

async def sum_numbers(numbers: List[int] = Body(...)):

return sum(numbers)

import time

from typing import Callable

from fastapi import APIRouter, FastAPI

from fastapi.routing import APIRoute

from starlette.requests import Request

from starlette.responses import Response

class TimedRoute(APIRoute):

def get_route_handler(self) -> Callable:

original_route_handler = super().get_route_handler()

async def custom_route_handler(request: Request) -> Response:

before = time.time()

response: Response = await original_route_handler(request)

duration = time.time() - before

response.headers["X-Response-Time"] = str(duration)

print(f"route duration: {duration}")

print(f"route response: {response}")

print(f"route response headers: {response.headers}")

return response

return custom_route_handler

app = FastAPI()

router = APIRouter(route_class=TimedRoute)

async def not_timed():

return {"message": "Not timed"}

async def timed():

return {"message": "It's the time of my life"}

app.include_router(router)

In some cases, you may want to override the logic used by the `Request` and `APIRoute` classes.

In particular, this may be a good alternative to logic in a middleware.

For example, if you want to read or manipulate the request body before it is processed by your application.

!!! danger

This is an "advanced" feature.

If you are just starting with **FastAPI** you might want to skip this section.

Some use cases include:

* Converting non-JSON request bodies to JSON (e.g. [`msgpack`](https://msgpack.org/index.html)).

* Decompressing gzip-compressed request bodies.

* Automatically logging all request bodies.

* Accessing the request body in an exception handler.

Let's see how to make use of a custom `Request` subclass to decompress gzip requests.

And an `APIRoute` subclass to use that custom request class.

First, we create a `GzipRequest` class, which will overwrite the `Request.body()` method to decompress the body in the presence of an appropriate header.

If there's no `gzip` in the header, it will not try to decompress the body.

That way, the same route class can handle gzip compressed or uncompressed requests.

```Python hl_lines="10 11 12 13 14 15 16 17"

{!./src/custom_request_and_route/tutorial001.py!}

Next, we create a custom subclass of `fastapi.routing.APIRoute` that will make use of the `GzipRequest`.

This time, it will overwrite the method `APIRoute.get_route_handler()`.

This method returns a function. And that function is what will receive a request and return a response.

Here we use it to create a `GzipRequest` from the original request.

```Python hl_lines="20 21 22 23 24 25 26 27 28"

{!./src/custom_request_and_route/tutorial001.py!}

!!! note "Technical Details"

A `Request` has a `request.scope` attribute, that's just a Python `dict` containing the metadata related to the request.

A `Request` also has a `request.receive`, that's a function to "receive" the body of the request.

The `scope` `dict` and `receive` function are both part of the ASGI specification.

And those two things, `scope` and `receive`, are what is needed to create a new `Request` instance.

To learn more about the `Request` check <a href="https://www.starlette.io/requests/" target="_blank">Starlette's docs about Requests</a>.

The only thing the function returned by `GzipRequest.get_route_handler` does differently is convert the `Request` to a `GzipRequest`.

Doing this, our `GzipRequest` will take care of decompressing the data (if necessary) before passing it to our *path operations*.

After that, all of the processing logic is the same.

But because of our changes in `GzipRequest.body`, the request body will be automatically decompressed when it is loaded by **FastAPI** when needed.

We can also use this same approach to access the request body in an exception handler.

All we need to do is handle the request inside a `try`/`except` block:

```Python hl_lines="15 17"

{!./src/custom_request_and_route/tutorial002.py!}

If an exception occurs, the`Request` instance will still be in scope, so we can read and make use of the request body when handling the error:

```Python hl_lines="18 19 20"

{!./src/custom_request_and_route/tutorial002.py!}

You can also set the `route_class` parameter of an `APIRouter`:

```Python hl_lines="25"

{!./src/custom_request_and_route/tutorial003.py!}

In this example, the *path operations* under the `router` will use the custom `TimedRoute` class, and will have an extra `X-Response-Time` header in the response with the time it took to generate the response:

```Python hl_lines="15 16 17 18 19"

{!./src/custom_request_and_route/tutorial003.py!}

def get_request_handler(

self.app = request_response(self.get_route_handler())

def get_route_handler(self) -> Callable:

return get_request_handler(

dependant=self.dependant,

body_field=self.body_field,

status_code=self.status_code,

response_class=self.response_class or JSONResponse,

response_field=self.secure_cloned_response_field,

response_model_include=self.response_model_include,

response_model_exclude=self.response_model_exclude,

response_model_by_alias=self.response_model_by_alias,

response_model_skip_defaults=self.response_model_skip_defaults,

dependency_overrides_provider=self.dependency_overrides_provider,

- Custom Request and APIRoute class: 'tutorial/custom-request-and-route.md'

import gzip

import json

import pytest

from starlette.requests import Request

from starlette.testclient import TestClient

from custom_request_and_route.tutorial001 import app

async def check_gzip_request(request: Request):

return {"request_class": type(request).__name__}

client = TestClient(app)

def test_gzip_request(compress):

n = 1000

headers = {}

body = [1] * n

data = json.dumps(body).encode()

if compress:

data = gzip.compress(data)

headers["Content-Encoding"] = "gzip"

response = client.post("/sum", data=data, headers=headers)

assert response.json() == {"sum": n}

def test_request_class():

response = client.get("/check-class")

assert response.json() == {"request_class": "GzipRequest"}

from starlette.testclient import TestClient

from custom_request_and_route.tutorial002 import app

client = TestClient(app)

def test_endpoint_works():

response = client.post("/", json=[1, 2, 3])

assert response.json() == 6

def test_exception_handler_body_access():

response = client.post("/", json={"numbers": [1, 2, 3]})

assert response.json() == {

"detail": {

"body": '{"numbers": [1, 2, 3]}',

"errors": [

{

"loc": ["body", "numbers"],

"msg": "value is not a valid list",

"type": "type_error.list",

}

],

}

}

from starlette.testclient import TestClient

from custom_request_and_route.tutorial003 import app

client = TestClient(app)

def test_get():

response = client.get("/")

assert response.json() == {"message": "Not timed"}

assert "X-Response-Time" not in response.headers

def test_get_timed():

response = client.get("/timed")

assert response.json() == {"message": "It's the time of my life"}

assert "X-Response-Time" in response.headers

assert float(response.headers["X-Response-Time"]) > 0