自定义请求和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属性，它只是一个包含与请求相关的元数据的Python dict。
一个Request还有一个request.receive，它是一个用于“接收”请求体的函数。
scope dict和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)