Clase personalizada de Request y APIRoute¶
En algunos casos, puede que quieras sobrescribir la lógica utilizada por las clases Request y APIRoute.
En particular, esta puede ser una buena alternativa a la lógica en un middleware.
Por ejemplo, si quieres leer o manipular el request body antes de que sea procesado por tu aplicación.
Advertencia
Esta es una funcionalidad "avanzada".
Si apenas estás comenzando con FastAPI, quizás quieras saltar esta sección.
Casos de uso¶
Algunos casos de uso incluyen:
- Convertir cuerpos de requests no-JSON a JSON (por ejemplo,
msgpack). - Descomprimir cuerpos de requests comprimidos con gzip.
- Registrar automáticamente todos los request bodies.
Manejo de codificaciones personalizadas de request body¶
Veamos cómo hacer uso de una subclase personalizada de Request para descomprimir requests gzip.
Y una subclase de APIRoute para usar esa clase de request personalizada.
Crear una clase personalizada GzipRequest¶
Consejo
Este es un ejemplo sencillo para demostrar cómo funciona. Si necesitas soporte para Gzip, puedes usar el GzipMiddleware proporcionado.
Primero, creamos una clase GzipRequest, que sobrescribirá el método Request.body() para descomprimir el cuerpo si hay un header apropiado.
Si no hay gzip en el header, no intentará descomprimir el cuerpo.
De esa manera, la misma clase de ruta puede manejar requests comprimidos con gzip o no comprimidos.
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)}
Crear una clase personalizada GzipRoute¶
A continuación, creamos una subclase personalizada de fastapi.routing.APIRoute que hará uso de GzipRequest.
Esta vez, sobrescribirá el método APIRoute.get_route_handler().
Este método devuelve una función. Y esa función es la que recibirá un request y devolverá un response.
Aquí lo usamos para crear un GzipRequest a partir del request 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)}
Detalles técnicos
Un Request tiene un atributo request.scope, que es simplemente un dict de Python que contiene los metadatos relacionados con el request.
Un Request también tiene un request.receive, que es una función para "recibir" el cuerpo del request.
El dict scope y la función receive son ambos parte de la especificación ASGI.
Y esas dos cosas, scope y receive, son lo que se necesita para crear una nueva Request instance.
Para aprender más sobre el Request, revisa la documentación de Starlette sobre Requests.
La única cosa que la función devuelta por GzipRequest.get_route_handler hace diferente es convertir el Request en un GzipRequest.
Haciendo esto, nuestro GzipRequest se encargará de descomprimir los datos (si es necesario) antes de pasarlos a nuestras path operations.
Después de eso, toda la lógica de procesamiento es la misma.
Pero debido a nuestros cambios en GzipRequest.body, el request body se descomprimirá automáticamente cuando sea cargado por FastAPI si es necesario.
Accediendo al request body en un manejador de excepciones¶
Consejo
Para resolver este mismo problema, probablemente sea mucho más fácil usar el body en un manejador personalizado para RequestValidationError (Manejo de Errores).
Pero este ejemplo sigue siendo válido y muestra cómo interactuar con los componentes internos.
También podemos usar este mismo enfoque para acceder al request body en un manejador de excepciones.
Todo lo que necesitamos hacer es manejar el request dentro de un bloque 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)
Si ocurre una excepción, la Request instance aún estará en el alcance, así que podemos leer y hacer uso del request body cuando manejamos el error:
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)
Clase personalizada APIRoute en un router¶
También puedes establecer el parámetro route_class de un 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)
En este ejemplo, las path operations bajo el router usarán la clase personalizada TimedRoute, y tendrán un header X-Response-Time extra en el response con el tiempo que tomó generar el response:
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)