Coverage for docs_src / custom_request_and_route / tutorial001_an_py310.py: 100%

1import gzip 6 ctx1abcdef

2from collections.abc import Callable 6 ctx1abcdef

3from typing import Annotated 6 ctx1abcdef

4 

5from fastapi import Body, FastAPI, Request, Response 6 ctx1abcdef

6from fastapi.routing import APIRoute 6 ctx1abcdef

7 

8 

9class GzipRequest(Request): 6 ctx1abcdef

10 async def body(self) -> bytes: 6 ctx1abcdef

11 if not hasattr(self, "_body"): 7 ctx1kghlimj

12 body = await super().body() 7 ctx1kghlimj

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

14 body = gzip.decompress(body) 4 ctx1ghij

15 self._body = body 7 ctx1kghlimj

16 return self._body 7 ctx1kghlimj

17 

18 

19class GzipRoute(APIRoute): 6 ctx1abcdef

20 def get_route_handler(self) -> Callable: 6 ctx1abcdef

21 original_route_handler = super().get_route_handler() 10 ctx1abncdeopfq

22 

23 async def custom_route_handler(request: Request) -> Response: 10 ctx1abncdeopfq

24 request = GzipRequest(request.scope, request.receive) 10 ctx1kgrhlismjt

25 return await original_route_handler(request) 10 ctx1kgrhlismjt

26 

27 return custom_route_handler 10 ctx1abncdeopfq

28 

29 

30app = FastAPI() 6 ctx1abcdef

31app.router.route_class = GzipRoute 6 ctx1abcdef

32 

33 

34@app.post("/sum") 6 ctx1abcdef

35async def sum_numbers(numbers: Annotated[list[int], Body()]): 6 ctx1abcdef

36 return {"sum": sum(numbers)} 7 ctx1kghlimj