Clase Request y APIRoute personalizada¶
En algunos casos, puede que quieras sobrescribir la lógica usada 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 cuerpo de la petición antes de que sea procesado por tu aplicación.
Peligro
Esta es una función "avanzada".
Si estás empezando con FastAPI puede que quieras saltarte esta sección.
Casos de uso¶
Algunos casos de uso incluyen:
- Convertir cuerpos de petición que no son JSON a JSON (por ejemplo,
msgpack). - Descomprimir cuerpos de petición comprimidos con gzip.
- Registrar automáticamente todos los cuerpos de petición.
Manejar codificaciones personalizadas del cuerpo de la petición¶
Veamos cómo hacer uso de una subclase personalizada de Request para descomprimir peticiones gzip.
Y una subclase de APIRoute para usar esa clase de petición personalizada.
Crear una clase GzipRequest personalizada¶
Consejo
Este es un ejemplo de juguete para demostrar cómo funciona, si necesitas soporte Gzip, puedes usar el GzipMiddleware proporcionado.
Primero, creamos una clase GzipRequest, que sobrescribirá el método Request.body() para descomprimir el cuerpo en presencia de una cabecera apropiada.
Si no hay gzip en la cabecera, no intentará descomprimir el cuerpo.
De esa manera, la misma clase de ruta puede manejar peticiones comprimidas con gzip o no comprimidas.
import gzip
from collections.abc import Callable
from typing import Annotated
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: Annotated[list[int], Body()]):
return {"sum": sum(numbers)}
🤓 Otras versiones y variantes
Consejo
Preferible usar la versión con Annotated si es posible.
import gzip
from collections.abc import Callable
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 GzipRoute personalizada¶
Luego, creamos una subclase personalizada de fastapi.routing.APIRoute que hará uso del GzipRequest.
Esta vez, sobrescribirá el método APIRoute.get_route_handler().
Este método devuelve una función. Y esa función es lo que recibirá una petición y devolverá una respuesta.
Aquí lo usamos para crear un GzipRequest a partir de la petición original.
import gzip
from collections.abc import Callable
from typing import Annotated
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: Annotated[list[int], Body()]):
return {"sum": sum(numbers)}
🤓 Otras versiones y variantes
Consejo
Preferible usar la versión con Annotated si es posible.
import gzip
from collections.abc import Callable
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 la petición.
Un Request también tiene un request.receive, que es una función para "recibir" el cuerpo de la petición.
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 instancia de Request.
Para aprender más sobre el Request consulta la documentación de Starlette sobre Requests.
Lo único que hace diferente la función devuelta por GzipRequest.get_route_handler 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 cuerpo de la petición será descomprimido automáticamente cuando sea cargado por FastAPI cuando sea necesario.
Acceder al cuerpo de la petición 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 cuerpo de la petición en un manejador de excepciones.
Todo lo que necesitamos hacer es manejar la petición dentro de un bloque try/except:
from collections.abc import Callable
from typing import Annotated
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: Annotated[list[int], Body()]):
return sum(numbers)
🤓 Otras versiones y variantes
Consejo
Preferible usar la versión con Annotated si es posible.
from collections.abc import Callable
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 instancia de Request seguirá en el scope, por lo que podemos leer y hacer uso del cuerpo de la petición al manejar el error:
from collections.abc import Callable
from typing import Annotated
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: Annotated[list[int], Body()]):
return sum(numbers)
🤓 Otras versiones y variantes
Consejo
Preferible usar la versión con Annotated si es posible.
from collections.abc import Callable
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 APIRoute personalizada en un router¶
También puedes establecer el parámetro route_class de un APIRouter:
import time
from collections.abc 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 una cabecera extra X-Response-Time en la respuesta con el tiempo que tomó generar la respuesta:
import time
from collections.abc 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)