自定义请求和APIRoute类¶
在某些情况下，你可能需要覆盖Request和APIRoute类使用的逻辑。
特别是，这可能是中间件逻辑的一个很好的替代方案。
例如，如果你想在应用程序处理请求体之前读取或操作它。
Danger
这是一个"高级"功能。
如果你刚开始使用FastAPI，可以跳过本节。
使用场景¶
一些使用场景包括：
- 将非JSON请求体转换为JSON（例如
msgpack）。 - 解压gzip压缩的请求体。
- 自动记录所有请求体。
处理自定义请求体编码¶
让我们看看如何利用自定义的Request子类来解压gzip请求。
以及使用该自定义请求类的APIRoute子类。
创建自定义GzipRequest类¶
Tip
这是一个演示工作原理的简单示例，如果需要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属性，那只是一个包含请求相关元数据的Pythondict。
Request还有一个request.receive，这是一个用于"接收"请求体的函数。
scope dict和receive函数都是ASGI规范的一部分。
这两者scope和receive是创建新Request实例所需的。
要了解更多关于Request的信息，请查看Starlette关于Requests的文档。
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)