Clase personalizada de Request y APIRoute¶
🌐 Traducción por IA y humanos
Esta traducción fue hecha por IA guiada por humanos. 🤝
Podría tener errores al interpretar el significado original, o sonar poco natural, etc. 🤖
Puedes mejorar esta traducciónayudándonos a guiar mejor al LLM de IA.
En algunos casos, puede que quieras sobrescribir la lógica utilizada por las clasesRequest yAPIRoute.
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 conFastAPI, 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 deRequest para descomprimir requests gzip.
Y una subclase deAPIRoute para usar esa clase de request personalizada.
Crear una clase personalizadaGzipRequest¶
Consejo
Este es un ejemplo sencillo para demostrar cómo funciona. Si necesitas soporte para Gzip, puedes usar elGzipMiddleware proporcionado.
Primero, creamos una claseGzipRequest, que sobrescribirá el métodoRequest.body() para descomprimir el cuerpo si hay un header apropiado.
Si no haygzip 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.
importgzipfromcollections.abcimportCallablefromtypingimportAnnotatedfromfastapiimportBody,FastAPI,Request,Responsefromfastapi.routingimportAPIRouteclassGzipRequest(Request):asyncdefbody(self)->bytes:ifnothasattr(self,"_body"):body=awaitsuper().body()if"gzip"inself.headers.getlist("Content-Encoding"):body=gzip.decompress(body)self._body=bodyreturnself._bodyclassGzipRoute(APIRoute):defget_route_handler(self)->Callable:original_route_handler=super().get_route_handler()asyncdefcustom_route_handler(request:Request)->Response:request=GzipRequest(request.scope,request.receive)returnawaitoriginal_route_handler(request)returncustom_route_handlerapp=FastAPI()app.router.route_class=GzipRoute@app.post("/sum")asyncdefsum_numbers(numbers:Annotated[list[int],Body()]):return{"sum":sum(numbers)}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
importgzipfromcollections.abcimportCallablefromfastapiimportBody,FastAPI,Request,Responsefromfastapi.routingimportAPIRouteclassGzipRequest(Request):asyncdefbody(self)->bytes:ifnothasattr(self,"_body"):body=awaitsuper().body()if"gzip"inself.headers.getlist("Content-Encoding"):body=gzip.decompress(body)self._body=bodyreturnself._bodyclassGzipRoute(APIRoute):defget_route_handler(self)->Callable:original_route_handler=super().get_route_handler()asyncdefcustom_route_handler(request:Request)->Response:request=GzipRequest(request.scope,request.receive)returnawaitoriginal_route_handler(request)returncustom_route_handlerapp=FastAPI()app.router.route_class=GzipRoute@app.post("/sum")asyncdefsum_numbers(numbers:list[int]=Body()):return{"sum":sum(numbers)}Crear una clase personalizadaGzipRoute¶
A continuación, creamos una subclase personalizada defastapi.routing.APIRoute que hará uso deGzipRequest.
Esta vez, sobrescribirá el métodoAPIRoute.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 unGzipRequest a partir del request original.
importgzipfromcollections.abcimportCallablefromtypingimportAnnotatedfromfastapiimportBody,FastAPI,Request,Responsefromfastapi.routingimportAPIRouteclassGzipRequest(Request):asyncdefbody(self)->bytes:ifnothasattr(self,"_body"):body=awaitsuper().body()if"gzip"inself.headers.getlist("Content-Encoding"):body=gzip.decompress(body)self._body=bodyreturnself._bodyclassGzipRoute(APIRoute):defget_route_handler(self)->Callable:original_route_handler=super().get_route_handler()asyncdefcustom_route_handler(request:Request)->Response:request=GzipRequest(request.scope,request.receive)returnawaitoriginal_route_handler(request)returncustom_route_handlerapp=FastAPI()app.router.route_class=GzipRoute@app.post("/sum")asyncdefsum_numbers(numbers:Annotated[list[int],Body()]):return{"sum":sum(numbers)}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
importgzipfromcollections.abcimportCallablefromfastapiimportBody,FastAPI,Request,Responsefromfastapi.routingimportAPIRouteclassGzipRequest(Request):asyncdefbody(self)->bytes:ifnothasattr(self,"_body"):body=awaitsuper().body()if"gzip"inself.headers.getlist("Content-Encoding"):body=gzip.decompress(body)self._body=bodyreturnself._bodyclassGzipRoute(APIRoute):defget_route_handler(self)->Callable:original_route_handler=super().get_route_handler()asyncdefcustom_route_handler(request:Request)->Response:request=GzipRequest(request.scope,request.receive)returnawaitoriginal_route_handler(request)returncustom_route_handlerapp=FastAPI()app.router.route_class=GzipRoute@app.post("/sum")asyncdefsum_numbers(numbers:list[int]=Body()):return{"sum":sum(numbers)}Detalles técnicos
UnRequest tiene un atributorequest.scope, que es simplemente undict de Python que contiene los metadatos relacionados con el request.
UnRequest también tiene unrequest.receive, que es una función para "recibir" el cuerpo del request.
Eldictscope y la funciónreceive son ambos parte de la especificación ASGI.
Y esas dos cosas,scope yreceive, son lo que se necesita para crear una nuevaRequest instance.
Para aprender más sobre elRequest, revisala documentación de Starlette sobre Requests.
La única cosa que la función devuelta porGzipRequest.get_route_handler hace diferente es convertir elRequest en unGzipRequest.
Haciendo esto, nuestroGzipRequest se encargará de descomprimir los datos (si es necesario) antes de pasarlos a nuestraspath operations.
Después de eso, toda la lógica de procesamiento es la misma.
Pero debido a nuestros cambios enGzipRequest.body, el request body se descomprimirá automáticamente cuando sea cargado porFastAPI 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 elbody en un manejador personalizado paraRequestValidationError (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 bloquetry/except:
fromcollections.abcimportCallablefromtypingimportAnnotatedfromfastapiimportBody,FastAPI,HTTPException,Request,Responsefromfastapi.exceptionsimportRequestValidationErrorfromfastapi.routingimportAPIRouteclassValidationErrorLoggingRoute(APIRoute):defget_route_handler(self)->Callable:original_route_handler=super().get_route_handler()asyncdefcustom_route_handler(request:Request)->Response:try:returnawaitoriginal_route_handler(request)exceptRequestValidationErrorasexc:body=awaitrequest.body()detail={"errors":exc.errors(),"body":body.decode()}raiseHTTPException(status_code=422,detail=detail)returncustom_route_handlerapp=FastAPI()app.router.route_class=ValidationErrorLoggingRoute@app.post("/")asyncdefsum_numbers(numbers:Annotated[list[int],Body()]):returnsum(numbers)🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromcollections.abcimportCallablefromfastapiimportBody,FastAPI,HTTPException,Request,Responsefromfastapi.exceptionsimportRequestValidationErrorfromfastapi.routingimportAPIRouteclassValidationErrorLoggingRoute(APIRoute):defget_route_handler(self)->Callable:original_route_handler=super().get_route_handler()asyncdefcustom_route_handler(request:Request)->Response:try:returnawaitoriginal_route_handler(request)exceptRequestValidationErrorasexc:body=awaitrequest.body()detail={"errors":exc.errors(),"body":body.decode()}raiseHTTPException(status_code=422,detail=detail)returncustom_route_handlerapp=FastAPI()app.router.route_class=ValidationErrorLoggingRoute@app.post("/")asyncdefsum_numbers(numbers:list[int]=Body()):returnsum(numbers)Si ocurre una excepción, laRequest instance aún estará en el alcance, así que podemos leer y hacer uso del request body cuando manejamos el error:
fromcollections.abcimportCallablefromtypingimportAnnotatedfromfastapiimportBody,FastAPI,HTTPException,Request,Responsefromfastapi.exceptionsimportRequestValidationErrorfromfastapi.routingimportAPIRouteclassValidationErrorLoggingRoute(APIRoute):defget_route_handler(self)->Callable:original_route_handler=super().get_route_handler()asyncdefcustom_route_handler(request:Request)->Response:try:returnawaitoriginal_route_handler(request)exceptRequestValidationErrorasexc:body=awaitrequest.body()detail={"errors":exc.errors(),"body":body.decode()}raiseHTTPException(status_code=422,detail=detail)returncustom_route_handlerapp=FastAPI()app.router.route_class=ValidationErrorLoggingRoute@app.post("/")asyncdefsum_numbers(numbers:Annotated[list[int],Body()]):returnsum(numbers)🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromcollections.abcimportCallablefromfastapiimportBody,FastAPI,HTTPException,Request,Responsefromfastapi.exceptionsimportRequestValidationErrorfromfastapi.routingimportAPIRouteclassValidationErrorLoggingRoute(APIRoute):defget_route_handler(self)->Callable:original_route_handler=super().get_route_handler()asyncdefcustom_route_handler(request:Request)->Response:try:returnawaitoriginal_route_handler(request)exceptRequestValidationErrorasexc:body=awaitrequest.body()detail={"errors":exc.errors(),"body":body.decode()}raiseHTTPException(status_code=422,detail=detail)returncustom_route_handlerapp=FastAPI()app.router.route_class=ValidationErrorLoggingRoute@app.post("/")asyncdefsum_numbers(numbers:list[int]=Body()):returnsum(numbers)Clase personalizadaAPIRoute en un router¶
También puedes establecer el parámetroroute_class de unAPIRouter:
importtimefromcollections.abcimportCallablefromfastapiimportAPIRouter,FastAPI,Request,Responsefromfastapi.routingimportAPIRouteclassTimedRoute(APIRoute):defget_route_handler(self)->Callable:original_route_handler=super().get_route_handler()asyncdefcustom_route_handler(request:Request)->Response:before=time.time()response:Response=awaitoriginal_route_handler(request)duration=time.time()-beforeresponse.headers["X-Response-Time"]=str(duration)print(f"route duration:{duration}")print(f"route response:{response}")print(f"route response headers:{response.headers}")returnresponsereturncustom_route_handlerapp=FastAPI()router=APIRouter(route_class=TimedRoute)@app.get("/")asyncdefnot_timed():return{"message":"Not timed"}@router.get("/timed")asyncdeftimed():return{"message":"It's the time of my life"}app.include_router(router)En este ejemplo, laspath operations bajo elrouter usarán la clase personalizadaTimedRoute, y tendrán un headerX-Response-Time extra en el response con el tiempo que tomó generar el response:
importtimefromcollections.abcimportCallablefromfastapiimportAPIRouter,FastAPI,Request,Responsefromfastapi.routingimportAPIRouteclassTimedRoute(APIRoute):defget_route_handler(self)->Callable:original_route_handler=super().get_route_handler()asyncdefcustom_route_handler(request:Request)->Response:before=time.time()response:Response=awaitoriginal_route_handler(request)duration=time.time()-beforeresponse.headers["X-Response-Time"]=str(duration)print(f"route duration:{duration}")print(f"route response:{response}")print(f"route response headers:{response.headers}")returnresponsereturncustom_route_handlerapp=FastAPI()router=APIRouter(route_class=TimedRoute)@app.get("/")asyncdefnot_timed():return{"message":"Not timed"}@router.get("/timed")asyncdeftimed():return{"message":"It's the time of my life"}app.include_router(router)