Personnaliser les classes Request et APIRoute¶
🌐 Traduction par IA et humains
Cette traduction a été réalisée par une IA guidée par des humains. 🤝
Elle peut contenir des erreurs d'interprétation du sens original, ou paraître peu naturelle, etc. 🤖
Vous pouvez améliorer cette traduction en nous aidant à mieux guider le LLM d'IA.
Dans certains cas, vous pouvez vouloir surcharger la logique utilisée par les classes Request et APIRoute.
En particulier, cela peut être une bonne alternative à une logique dans un middleware.
Par exemple, si vous voulez lire ou manipuler le corps de la requête avant qu'il ne soit traité par votre application.
Danger
Ceci est une fonctionnalité « avancée ».
Si vous débutez avec FastAPI, vous pouvez ignorer cette section.
Cas d'utilisation¶
Voici quelques cas d'utilisation :
- Convertir des corps de requête non JSON en JSON (par exemple
msgpack). - Décompresser des corps de requête compressés en gzip.
- Journaliser automatiquement tous les corps de requête.
Gérer les encodages personnalisés du corps de la requête¶
Voyons comment utiliser une sous-classe personnalisée de Request pour décompresser des requêtes gzip.
Et une sous-classe d'APIRoute pour utiliser cette classe de requête personnalisée.
Créer une classe GzipRequest personnalisée¶
Astuce
Il s'agit d'un exemple simplifié pour montrer le fonctionnement ; si vous avez besoin de la prise en charge de Gzip, vous pouvez utiliser le GzipMiddleware fourni.
Commencez par créer une classe GzipRequest, qui va surcharger la méthode Request.body() pour décompresser le corps en présence d'un en-tête approprié.
S'il n'y a pas gzip dans l'en-tête, elle n'essaiera pas de décompresser le corps.
De cette manière, la même classe de route peut gérer des requêtes gzip compressées ou non compressées.
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)}
🤓 Other versions and variants
Tip
Prefer to use the Annotated version if possible.
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)}
Créer une classe GzipRoute personnalisée¶
Ensuite, nous créons une sous-classe personnalisée de fastapi.routing.APIRoute qui utilisera GzipRequest.
Cette fois, elle va surcharger la méthode APIRoute.get_route_handler().
Cette méthode renvoie une fonction. Et c'est cette fonction qui recevra une requête et retournera une réponse.
Ici, nous l'utilisons pour créer une GzipRequest à partir de la requête originale.
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)}
🤓 Other versions and variants
Tip
Prefer to use the Annotated version if possible.
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)}
Détails techniques
Un Request possède un attribut request.scope, qui n'est qu'un dict Python contenant les métadonnées liées à la requête.
Un Request a également un request.receive, qui est une fonction pour « recevoir » le corps de la requête.
Le dict scope et la fonction receive font tous deux partie de la spécification ASGI.
Et ces deux éléments, scope et receive, sont ce dont on a besoin pour créer une nouvelle instance de Request.
Pour en savoir plus sur Request, consultez la documentation de Starlette sur les requêtes.
La seule chose que fait différemment la fonction renvoyée par GzipRequest.get_route_handler, c'est de convertir la Request en GzipRequest.
Ce faisant, notre GzipRequest se chargera de décompresser les données (si nécessaire) avant de les transmettre à nos chemins d'accès.
Après cela, toute la logique de traitement est identique.
Mais grâce à nos modifications dans GzipRequest.body, le corps de la requête sera automatiquement décompressé lorsque FastAPI le chargera, si nécessaire.
Accéder au corps de la requête dans un gestionnaire d'exceptions¶
Astuce
Pour résoudre ce même problème, il est probablement beaucoup plus simple d'utiliser body dans un gestionnaire personnalisé pour RequestValidationError (Gérer les erreurs).
Mais cet exemple reste valable et montre comment interagir avec les composants internes.
Nous pouvons également utiliser cette même approche pour accéder au corps de la requête dans un gestionnaire d'exceptions.
Il suffit de traiter la requête dans un bloc 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)
🤓 Other versions and variants
Tip
Prefer to use the Annotated version if possible.
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 une exception se produit, l'instance de Request sera toujours dans la portée, ce qui nous permet de lire et d'utiliser le corps de la requête lors du traitement de l'erreur :
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)
🤓 Other versions and variants
Tip
Prefer to use the Annotated version if possible.
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)
Utiliser une classe APIRoute personnalisée dans un routeur¶
Vous pouvez également définir le paramètre route_class d'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)
Dans cet exemple, les chemins d'accès sous le router utiliseront la classe personnalisée TimedRoute, et auront un en-tête supplémentaire X-Response-Time dans la réponse avec le temps nécessaire pour générer la réponse :
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)