自定义请求和 APIRoute 类¶
在某些情况下，您可能希望覆盖 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。
Request 还有一个 request.receive，它是一个用于“接收”请求体的函数。
scope 字典和 receive 函数都是 ASGI 规范的一部分。
而这两样东西，scope 和 receive，是创建一个新的 Request 实例所需要的。
要了解有关 Request 的更多信息，请查看 Starlette 关于请求的文档。
GzipRequest.get_route_handler 返回的函数所做的唯一不同之处是将 Request 转换为 GzipRequest。
这样做，我们的 GzipRequest 将在将数据传递给我们的*路径操作*之前负责解压数据（如果需要）。
在那之后，所有的处理逻辑都是相同的。
但是由于我们在 GzipRequest.body 中做了更改，当 FastAPI 需要时，请求体将在加载时被自动解压。
在异常处理程序中访问请求体¶
我们也可以使用同样的方法在异常处理程序中访问请求体。
我们所需要做的就是在 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 类¶
您还可以设置 APIRouter 的 route_class 参数
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)