Benutzerdefinierte Request- und APIRoute-Klasse¶
In einigen Fällen möchten Sie möglicherweise die von den Klassen Request
und APIRoute
verwendete Logik überschreiben.
Das kann insbesondere eine gute Alternative zur Logik in einer Middleware sein.
Wenn Sie beispielsweise den Requestbody lesen oder manipulieren möchten, bevor er von Ihrer Anwendung verarbeitet wird.
"Gefahr"
Dies ist eine „fortgeschrittene“ Funktion.
Wenn Sie gerade erst mit FastAPI beginnen, möchten Sie diesen Abschnitt vielleicht überspringen.
Anwendungsfälle¶
Einige Anwendungsfälle sind:
- Konvertieren von Nicht-JSON-Requestbodys nach JSON (z. B.
msgpack
). - Dekomprimierung gzip-komprimierter Requestbodys.
- Automatisches Loggen aller Requestbodys.
Handhaben von benutzerdefinierten Requestbody-Kodierungen¶
Sehen wir uns an, wie Sie eine benutzerdefinierte Request
-Unterklasse verwenden, um gzip-Requests zu dekomprimieren.
Und eine APIRoute
-Unterklasse zur Verwendung dieser benutzerdefinierten Requestklasse.
Eine benutzerdefinierte GzipRequest
-Klasse erstellen¶
"Tipp"
Dies ist nur ein einfaches Beispiel, um zu demonstrieren, wie es funktioniert. Wenn Sie Gzip-Unterstützung benötigen, können Sie die bereitgestellte GzipMiddleware
verwenden.
Zuerst erstellen wir eine GzipRequest
-Klasse, welche die Methode Request.body()
überschreibt, um den Body bei Vorhandensein eines entsprechenden Headers zu dekomprimieren.
Wenn der Header kein gzip
enthält, wird nicht versucht, den Body zu dekomprimieren.
Auf diese Weise kann dieselbe Routenklasse gzip-komprimierte oder unkomprimierte Requests verarbeiten.
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)}
Eine benutzerdefinierte GzipRoute
-Klasse erstellen¶
Als Nächstes erstellen wir eine benutzerdefinierte Unterklasse von fastapi.routing.APIRoute
, welche GzipRequest
nutzt.
Dieses Mal wird die Methode APIRoute.get_route_handler()
überschrieben.
Diese Methode gibt eine Funktion zurück. Und diese Funktion empfängt einen Request und gibt eine Response zurück.
Hier verwenden wir sie, um aus dem ursprünglichen Request einen GzipRequest
zu erstellen.
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)}
"Technische Details"
Ein Request
hat ein request.scope
-Attribut, welches einfach ein Python-dict
ist, welches die mit dem Request verbundenen Metadaten enthält.
Ein Request
hat auch ein request.receive
, welches eine Funktion ist, die den Hauptteil des Requests empfängt.
Das scope
-dict
und die receive
-Funktion sind beide Teil der ASGI-Spezifikation.
Und diese beiden Dinge, scope
und receive
, werden benötigt, um eine neue Request
-Instanz zu erstellen.
Um mehr über den Request
zu erfahren, schauen Sie sich Starlettes Dokumentation zu Requests an.
Das Einzige, was die von GzipRequest.get_route_handler
zurückgegebene Funktion anders macht, ist die Konvertierung von Request
in ein GzipRequest
.
Dabei kümmert sich unser GzipRequest
um die Dekomprimierung der Daten (falls erforderlich), bevor diese an unsere Pfadoperationen weitergegeben werden.
Danach ist die gesamte Verarbeitungslogik dieselbe.
Aufgrund unserer Änderungen in GzipRequest.body
wird der Requestbody jedoch bei Bedarf automatisch dekomprimiert, wenn er von FastAPI geladen wird.
Zugriff auf den Requestbody in einem Exceptionhandler¶
"Tipp"
Um dasselbe Problem zu lösen, ist es wahrscheinlich viel einfacher, den body
in einem benutzerdefinierten Handler für RequestValidationError
zu verwenden (Fehlerbehandlung).
Dieses Beispiel ist jedoch immer noch gültig und zeigt, wie mit den internen Komponenten interagiert wird.
Wir können denselben Ansatz auch verwenden, um in einem Exceptionhandler auf den Requestbody zuzugreifen.
Alles, was wir tun müssen, ist, den Request innerhalb eines try
/except
-Blocks zu handhaben:
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)
Wenn eine Exception auftritt, befindet sich die Request
-Instanz weiterhin im Gültigkeitsbereich, sodass wir den Requestbody lesen und bei der Fehlerbehandlung verwenden können:
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)
Benutzerdefinierte APIRoute
-Klasse in einem Router¶
Sie können auch den Parameter route_class
eines APIRouter
festlegen:
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)
In diesem Beispiel verwenden die Pfadoperationen unter dem router
die benutzerdefinierte TimedRoute
-Klasse und haben in der Response einen zusätzlichen X-Response-Time
-Header mit der Zeit, die zum Generieren der Response benötigt wurde:
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)