在某些情况下,您可能想覆盖
Request
和
APIRoute
类
使用的逻辑
。
特别是,这可能是中间件中逻辑的一个很好的选择。
例如,如果您想在应用程序处理请求主体之前读取或操纵该请求主体。
危险
这是“高级”功能。
如果您只是从 FastAPI 开始, 则 可能要跳过本节。
一些用例包括:
msgpack
)。
让我们看看如何利用自定义
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)
文章来源互联网,如有侵权,请联系管理员删除。邮箱:417803890@qq.com / QQ:417803890
Python Free
邮箱:417803890@qq.com
QQ:417803890
皖ICP备19001818号-4
© 2019 copyright www.pythonf.cn - All rights reserved
微信扫一扫关注公众号:
Python Free