在某些情況下，您可能想覆蓋 Request 和 APIRoute 類 使用的邏輯 。
特別是，這可能是中間件中邏輯的一個很好的選擇。
例如，如果您想在應用程序處理請求主體之前讀取或操縱該請求主體。
危險
這是“高級”功能。
如果您只是從 FastAPI 開始， 則 可能要跳過本節。
用例
一些用例包括：
- 將非JSON請求正文轉換為JSON（例如
msgpack）。 - 解壓縮gzip壓縮的請求正文。
- 自動記錄所有請求正文。
處理自定義請求主體編碼
讓我們看看如何利用自定義 Request 子類解壓縮gzip請求。
還有一個 APIRoute 使用該自定義請求類 的 子類。
創建自定義 GzipRequest 類
小費
這是一個玩具示例，用於演示其工作原理。如果需要Gzip支持，則可以使用提供的 GzipMiddleware 。
首先，我們創建一個 GzipRequest 類，該類將 Request.body() 在存在適當頭的情況下 覆蓋 用於解壓縮主體 的 方法。
如果 gzip 標題中 沒有 ，它將不會嘗試解壓縮主體。
這樣，相同的路由類可以處理gzip壓縮或未壓縮的請求。
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)}
創建自定義 GzipRoute 類
接下來，我們將創建一個自定義子類 fastapi.routing.APIRoute ，以使用 GzipRequest 。
這次，它將覆蓋方法 APIRoute.get_route_handler() 。
此方法返回一個函數。 該函數將接收請求並返回響應。
在這里，我們使用它 GzipRequest 從原始請求中 創建一個 。
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)}
技術細節
一個 Request 具有 request.scope 屬性，這只是一個Python dict 包含相關請求的元數據。
A Request 也有一個 request.receive ，即“接收”請求正文的功能。
該 scope dict 和 receive 功能是ASGI技術規格的一部分。
而這兩個 scope 和 receive 是創建新 Request 實例 所需的 。
要了解有關 Request 檢查 Starlette的關於Requests的文檔的 更多信息 。
函數返回的唯一 GzipRequest.get_route_handler 不同之處是將轉換 Request 為 GzipRequest 。
這樣做，我們 GzipRequest 將在將數據傳遞給我們的 path操作 之前，對數據進行解壓縮（如果需要） 。
之后，所有處理邏輯都是相同的。
但是由於我們的更改 GzipRequest.body ， 在需要時 由 FastAPI 加載請求主體時，請求主體將自動解壓縮 。
在異常處理程序中訪問請求主體
小費
為了解決這個問題， body 在 RequestValidationError （ 處理錯誤 ） 的自定義處理程序中 使用可能要容易 得多 。
但是此示例仍然有效，並且顯示了如何與內部組件進行交互。
我們還可以使用相同的方法在異常處理程序中訪問請求正文。
我們需要做的就是在 try / except 塊中 處理請求 ：
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)
如果發生異常， Request 實例仍將在范圍內，因此在處理錯誤時我們可以讀取並利用請求正文：
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)
自定義 APIRoute 一個路由器類
您還可以設置 route_class 參數 APIRouter ：
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)
在這個例子中， 路徑操作 下， router 將使用自定義 TimedRoute 類，並有一個額外的 X-Response-Time 與它采取產生響應的時間響應頭：
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)
轉：https://www.pythonf.cn/read/56960