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:
También puedes establecer el parámetro route_class de un APIRouter:
importtimefromtypingimportCallablefromfastapiimportAPIRouter,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, 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:
importtimefromtypingimportCallablefromfastapiimportAPIRouter,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)