Requisições Personalizadas e Classes da APIRoute¶
Em algum casos, você pode querer sobreescrever a lógica usada pelas classes Requeste APIRoute.
Em particular, isso pode ser uma boa alternativa para uma lógica em um middleware
Por exemplo, se você quiser ler ou manipular o corpo da requisição antes que ele seja processado pela sua aplicação.
Perigo
Isso é um recurso "avançado".
Se você for um iniciante em FastAPI você deve considerar pular essa seção.
Casos de Uso¶
Alguns casos de uso incluem:
- Converter requisições não-JSON para JSON (por exemplo,
msgpack). - Descomprimir corpos de requisição comprimidos com gzip.
- Registrar automaticamente todos os corpos de requisição.
Manipulando codificações de corpo de requisição personalizadas¶
Vamos ver como usar uma subclasse personalizada de Request para descomprimir requisições gzip.
E uma subclasse de APIRoute para usar essa classe de requisição personalizada.
Criar uma classe GzipRequest personalizada¶
Dica
Isso é um exemplo de brincadeira para demonstrar como funciona, se você precisar de suporte para Gzip, você pode usar o GzipMiddleware fornecido.
Primeiro, criamos uma classe GzipRequest, que irá sobrescrever o método Request.body() para descomprimir o corpo na presença de um cabeçalho apropriado.
Se não houver gzip no cabeçalho, ele não tentará descomprimir o corpo.
Dessa forma, a mesma classe de rota pode lidar com requisições comprimidas ou não comprimidas.
import gzip
from typing import Callable, List
from fastapi import Body, FastAPI, Request, Response
from fastapi.routing import APIRoute
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
@app.post("/sum")
async def sum_numbers(numbers: List[int] = Body()):
return {"sum": sum(numbers)}
Criar uma classe GzipRoute personalizada¶
Em seguida, criamos uma subclasse personalizada de fastapi.routing.APIRoute que fará uso do GzipRequest.
Dessa vez, ele irá sobrescrever o método APIRoute.get_route_handler().
Esse método retorna uma função. E essa função é o que irá receber uma requisição e retornar uma resposta.
Aqui nós usamos para criar um GzipRequest a partir da requisição original.
import gzip
from typing import Callable, List
from fastapi import Body, FastAPI, Request, Response
from fastapi.routing import APIRoute
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
@app.post("/sum")
async def sum_numbers(numbers: List[int] = Body()):
return {"sum": sum(numbers)}
Detalhes Técnicos
Um Request também tem um request.receive, que é uma função para "receber" o corpo da requisição.
Um Request também tem um request.receive, que é uma função para "receber" o corpo da requisição.
O dicionário scope e a função receive são ambos parte da especificação ASGI.
E essas duas coisas, scope e receive, são o que é necessário para criar uma nova instância de Request.
Para aprender mais sobre o Request confira a documentação do Starlette sobre Requests.
A única coisa que a função retornada por GzipRequest.get_route_handler faz de diferente é converter o Request para um GzipRequest.
Fazendo isso, nosso GzipRequest irá cuidar de descomprimir os dados (se necessário) antes de passá-los para nossas operações de rota.
Depois disso, toda a lógica de processamento é a mesma.
Mas por causa das nossas mudanças em GzipRequest.body, o corpo da requisição será automaticamente descomprimido quando for carregado pelo FastAPI quando necessário.
Acessando o corpo da requisição em um manipulador de exceção¶
Dica
Para resolver esse mesmo problema, é provavelmente muito mais fácil usar o body em um manipulador personalizado para RequestValidationError (Tratando Erros).
Mas esse exemplo ainda é valido e mostra como interagir com os componentes internos.
Também podemos usar essa mesma abordagem para acessar o corpo da requisição em um manipulador de exceção.
Tudo que precisamos fazer é manipular a requisição dentro de um bloco try/except:
from typing import Callable, List
from fastapi import Body, FastAPI, HTTPException, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute
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
@app.post("/")
async def sum_numbers(numbers: List[int] = Body()):
return sum(numbers)
Se uma exceção ocorrer, a instância Request ainda estará em escopo, então podemos ler e fazer uso do corpo da requisição ao lidar com o erro:
from typing import Callable, List
from fastapi import Body, FastAPI, HTTPException, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute
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
@app.post("/")
async def sum_numbers(numbers: List[int] = Body()):
return sum(numbers)
Classe APIRoute personalizada em um router¶
você também pode definir o parametro route_class de uma APIRouter;
import time
from typing import Callable
from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute
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)
@app.get("/")
async def not_timed():
return {"message": "Not timed"}
@router.get("/timed")
async def timed():
return {"message": "It's the time of my life"}
app.include_router(router)
Nesse exemplo, as operações de rota sob o router irão usar a classe TimedRoute personalizada, e terão um cabeçalho extra X-Response-Time na resposta com o tempo que levou para gerar a resposta:
import time
from typing import Callable
from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute
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)
@app.get("/")
async def not_timed():
return {"message": "Not timed"}
@router.get("/timed")
async def timed():
return {"message": "It's the time of my life"}
app.include_router(router)