在某些情况下，您可能想覆盖 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