1 Request body重复消费简析
关于Request body重复消费读取的问题，其实在之前就有介绍到，但是之前浅尝辄止，管中窥豹，未能深入其中进行探究。近期刚好有些许时间再一探究竟。
2 重复消费的现象示例
重复消费的时候，从源码可以看得到我们的Request body 和 Request json其实都是一个协程函数，而不是一个属性值，所以我们的在进行Request body进行读取的时候，是需要进行相关的协程的调用如下示例代码：
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel
from starlette.requests import Request
from starlette.middleware.base import BaseHTTPMiddleware
app = FastAPI()
class LogerMiddleware1(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
_body = await request.body()
response = await call_next(request)
return response
class AuthMiddleware2(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
_body = await request.body()
response = await call_next(request)
return response
from fastapi import APIRouter
bp = APIRouter(tags=['权限管理'], prefix='/api/v1')
class AdminLogin(BaseModel):
"""
管理员登陆 提交的参数信息
"""
password: str = None
username: str = None
@bp.post("/one")
async def sum_numbers(req: Request):
print('one接口内的：', id(req))
return 'one'
@bp.post("/two")
async def sum_numbers(req: Request, sdsada: AdminLogin):
print('two接口内的：', id(req))
return 'two'
@app.post("/inapprsp")
async def inapprsp(req: Request, sdsada: AdminLogin):
print('inapprsp-接口内的：', id(req))
return 'inapprsp'
# 注册中间件
app.add_middleware(LogerMiddleware1)
app.add_middleware(AuthMiddleware2)
# 添加路由组
app.include_router(bp)
if __name__ == '__main__':
uvicorn.run(
app=app,
host="127.0.0.1",
port=5586,
workers=1,
reload=False,
debug=False,
)
示例代码中，我们定义了两个用户中间件，一个是LogerMiddleware1和AuthMiddleware2， 在中间件的内部都涉及到了boay的读取：
_body = await request.body()
此时启动服务的话，会无法启动，需要只能有一个中间件进行读取body。所以只能注释某一个中间件中的读取。注释之后，但是此时我们的随意访问一个路由的话，则会产生阻塞，一直等待。
但是最终测试发现，其实我们的之前案例其实是有问题，由始至终其实我们的中间内是无法直接的进行
_body = await request.body()
因为本身我们的在处理请求的时候，我们的路由处理的时候，其实也会读取消费
_body = await request.body()
所以也相当于如果我们的中间件已经消费了，但是到了路由处理层的时候，已经没有数据了，所以还是会造成无法读取的结果。
追溯到源码内部的，可以可以得到一些关于asgi协议的层上，首先我们的知道Fastapi本身是不提供运行容器，所以它依赖于unicorn等类型的容器来运行，Fastapi其实就是对asgi协议一层的封装扩展，追溯到一个简单的asgi的:
async def app(scope, receive, send):
# 一个最简单的ASGI应用程序
assert scope['type'] == 'http'
await receive()
await send({
'type': 'http.response.start',
'status': 200,
'headers': [
[b'content-type', b'text/plain'],
]
})
await send({
'type': 'http.response.body',
'body': b'nihao, world!',
})
其中里面的：
scope：类似于我们知道的wsgi协议中的environ，主要是包含请求连接具体信息 receive：异步可调用接口，用于收消息 send：异步可调用接口，用于发消息
对于是框架中的协议：
from starlette.types import ASGIApp, Receive, Scope, Send，Message
在深入到内部Route话，其实我们的可以看得到我们的每一个路由对象本身其实也是一个asgi协议实现的一个对象，
from starlette.routing import BaseRoute
查看BaseRoute源码可以分析到具体的实现：
class BaseRoute:
def matches(self, scope: Scope) -> typing.Tuple[Match, Scope]:
raise NotImplementedError() # pragma: no cover
def url_path_for(self, name: str, **path_params: str) -> URLPath:
raise NotImplementedError() # pragma: no cover
async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
raise NotImplementedError() # pragma: no cover
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
"""
A route may be used in isolation as a stand-alone ASGI app.
This is a somewhat contrived case, as they'll almost always be used
within a Router, but could be useful for some tooling and minimal apps.
"""
match, child_scope = self.matches(scope)
if match == Match.NONE:
if scope["type"] == "http":
response = PlainTextResponse("Not Found", status_code=404)
await response(scope, receive, send)
elif scope["type"] == "websocket":
websocket_close = WebSocketClose()
await websocket_close(scope, receive, send)
return
scope.update(child_scope)
await self.handle(scope, receive, send)
所以它也是一个标准asgi协议实现。但是其中我们的receive就是包含了我们的请求信息的之一，因为的Request就是通过scope: Scope, receive: Receive,来封装实现的，之前我们的也验证过在不同的中间件中我们的Request本身就是一个不同的对象。
在看我们的route的源码：
class Router:
def __init__(
self,
routes: typing.Sequence[BaseRoute] = None,
redirect_slashes: bool = True,
default: ASGIApp = None,
on_startup: typing.Sequence[typing.Callable] = None,
on_shutdown: typing.Sequence[typing.Callable] = None,
lifespan: typing.Callable[[typing.Any], typing.AsyncContextManager] = None,
) -> None:
self.routes = [] if routes is None else list(routes)
self.redirect_slashes = redirect_slashes
self.default = self.not_found if default is None else default
self.on_startup = [] if on_startup is None else list(on_startup)
self.on_shutdown = [] if on_shutdown is None else list(on_shutdown)
if lifespan is None:
self.lifespan_context: typing.Callable[
[typing.Any], typing.AsyncContextManager
] = _DefaultLifespan(self)
elif inspect.isasyncgenfunction(lifespan):
warnings.warn(
"async generator function lifespans are deprecated, "
"use an @contextlib.asynccontextmanager function instead",
DeprecationWarning,
)
self.lifespan_context = asynccontextmanager(
lifespan, # type: ignore[arg-type]
)
elif inspect.isgeneratorfunction(lifespan):
warnings.warn(
"generator function lifespans are deprecated, "
"use an @contextlib.asynccontextmanager function instead",
DeprecationWarning,
)
self.lifespan_context = _wrap_gen_lifespan_context(
lifespan, # type: ignore[arg-type]
)
else:
self.lifespan_context = lifespan
async def not_found(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] == "websocket":
websocket_close = WebSocketClose()
await websocket_close(scope, receive, send)
return
# If we're running inside a starlette application then raise an
# exception, so that the configurable exception handler can deal with
# returning the response. For plain ASGI apps, just return the response.
if "app" in scope:
raise HTTPException(status_code=404)
else:
response = PlainTextResponse("Not Found", status_code=404)
await response(scope, receive, send)
def url_path_for(self, name: str, **path_params: str) -> URLPath:
for route in self.routes:
try:
return route.url_path_for(name, **path_params)
except NoMatchFound:
pass
raise NoMatchFound()
async def startup(self) -> None:
"""
Run any `.on_startup` event handlers.
"""
for handler in self.on_startup:
if asyncio.iscoroutinefunction(handler):
await handler()
else:
handler()
async def shutdown(self) -> None:
"""
Run any `.on_shutdown` event handlers.
"""
for handler in self.on_shutdown:
if asyncio.iscoroutinefunction(handler):
await handler()
else:
handler()
async def lifespan(self, scope: Scope, receive: Receive, send: Send) -> None:
"""
Handle ASGI lifespan messages, which allows us to manage application
startup and shutdown events.
"""
started = False
app = scope.get("app")
await receive()
try:
async with self.lifespan_context(app):
await send({"type": "lifespan.startup.complete"})
started = True
await receive()
except BaseException:
exc_text = traceback.format_exc()
if started:
await send({"type": "lifespan.shutdown.failed", "message": exc_text})
else:
await send({"type": "lifespan.startup.failed", "message": exc_text})
raise
else:
await send({"type": "lifespan.shutdown.complete"})
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
"""
The main entry point to the Router class.
"""
assert scope["type"] in ("http", "websocket", "lifespan")
if "router" not in scope:
scope["router"] = self
if scope["type"] == "lifespan":
await self.lifespan(scope, receive, send)
return
partial = None
for route in self.routes:
# Determine if any route matches the incoming scope,
# and hand over to the matching route if found.
match, child_scope = route.matches(scope)
if match == Match.FULL:
scope.update(child_scope)
await route.handle(scope, receive, send)
return
elif match == Match.PARTIAL and partial is None:
partial = route
partial_scope = child_scope
if partial is not None:
# Handle partial matches. These are cases where an endpoint is
# able to handle the request, but is not a preferred option.
# We use this in particular to deal with "405 Method Not Allowed".
scope.update(partial_scope)
await partial.handle(scope, receive, send)
return
if scope["type"] == "http" and self.redirect_slashes and scope["path"] != "/":
redirect_scope = dict(scope)
if scope["path"].endswith("/"):
redirect_scope["path"] = redirect_scope["path"].rstrip("/")
else:
redirect_scope["path"] = redirect_scope["path"] + "/"
for route in self.routes:
match, child_scope = route.matches(redirect_scope)
if match != Match.NONE:
redirect_url = URL(scope=redirect_scope)
response = RedirectResponse(url=str(redirect_url))
await response(scope, receive, send)
return
await self.default(scope, receive, send)
def __eq__(self, other: typing.Any) -> bool:
return isinstance(other, Router) and self.routes == other.routes
# The following usages are now discouraged in favour of configuration
# during Router.__init__(...)
def mount(self, path: str, app: ASGIApp, name: str = None) -> None:
route = Mount(path, app=app, name=name)
self.routes.append(route)
def host(self, host: str, app: ASGIApp, name: str = None) -> None:
route = Host(host, app=app, name=name)
self.routes.append(route)
def add_route(
self,
path: str,
endpoint: typing.Callable,
methods: typing.List[str] = None,
name: str = None,
include_in_schema: bool = True,
) -> None:
route = Route(
path,
endpoint=endpoint,
methods=methods,
name=name,
include_in_schema=include_in_schema,
)
self.routes.append(route)
def add_websocket_route(
self, path: str, endpoint: typing.Callable, name: str = None
) -> None:
route = WebSocketRoute(path, endpoint=endpoint, name=name)
self.routes.append(route)
def route(
self,
path: str,
methods: typing.List[str] = None,
name: str = None,
include_in_schema: bool = True,
) -> typing.Callable:
def decorator(func: typing.Callable) -> typing.Callable:
self.add_route(
path,
func,
methods=methods,
name=name,
include_in_schema=include_in_schema,
)
return func
return decorator
def websocket_route(self, path: str, name: str = None) -> typing.Callable:
def decorator(func: typing.Callable) -> typing.Callable:
self.add_websocket_route(path, func, name=name)
return func
return decorator
def add_event_handler(self, event_type: str, func: typing.Callable) -> None:
assert event_type in ("startup", "shutdown")
if event_type == "startup":
self.on_startup.append(func)
else:
self.on_shutdown.append(func)
def on_event(self, event_type: str) -> typing.Callable:
def decorator(func: typing.Callable) -> typing.Callable:
self.add_event_handler(event_type, func)
return func
return decorator
所以可以看得到出，如果我们存在一个地方已经提交读取了的body的数据，相当于已经提前的从我们的一个请求队列读取了数据，读取完成后，我们的队列已经删除了数据了，此时第二地方在重复进行读取的时候，就会无法读取到数据，出于一直等待的状态。
3 多种解决方案
方案1：设置reque.state自定义路由类处理
通过自定义路由类，在自定义路由类中处理且只读取一次body数据，然后通过设置request.state给予一个新的属性值._body的方式来解决，但是这种方式只能在后续路由函数内进行调用获取，如果是还需要从中间内读取的话则还是无法读取到，因为路由是内部才去处理，，完整示例代码：
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel
from fastapi import APIRouter
from starlette.requests import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import Message, Receive, Scope, Send
app = FastAPI()
from fastapi.routing import APIRoute
from typing import Callable, List
from fastapi import Response
class BodySetRoute(APIRoute):
def get_route_handler(self) -> Callable:
original_route_handler = super().get_route_handler()
async def custom_route_handler(request: Request) -> Response:
request.state._body = await request.body()
# await proxy_get_body(request)
return await original_route_handler(request)
return custom_route_handler
app.router.route_class=BodySetRoute
class LogerMiddleware1(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# _body = await proxy_get_body(request)
response = await call_next(request)
if hasattr(request.state, '_body'):
print("第二次读取：", request.state._body)
return response
class AuthMiddleware2(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# _body = await proxy_get_body(request)
if hasattr(request.state,'_body'):
print("第1次读取：", request.state._body)
response = await call_next(request)
return response
class AdminLogin(BaseModel):
"""
管理员登陆 提交的参数信息
"""
password: str = None
username: str = None
@app.post("/inapprsp")
async def inapprsp(req: Request, sdsada: AdminLogin):
print('inapprsp-接口内的：', id(req))
print("路由内多次读取：", req.state._body)
print("路由内多次读取：", req.state._body)
print("路由内多次读取：", req.state._body)
print("路由内多次读取：", req.state._body)
return 'inapprsp'
# 注册中间件
app.add_middleware(LogerMiddleware1)
app.add_middleware(AuthMiddleware2)
if __name__ == '__main__':
uvicorn.run(
app='main2:app',
host="127.0.0.1",
port=5586,
reload=True,
)
或者在自定义路由类内自定一个代理读取
from starlette.types import Message
async def proxy_get_body(request: Request) -> bytes:
async def receive() -> Message:
return {"type": "http.request", "body": body}
body = await request.body()
request._receive = receive
return body
方案2：设新增缓存协程函数
class BodySetRoute(APIRoute):
def get_route_handler(self) -> Callable:
original_route_handler = super().get_route_handler()
async def custom_route_handler(request: Request) -> Response:
from starlette.types import Message
async def proxy_get_body(request: Request) -> bytes:
async def receive() -> Message:
return {"type": "http.request", "body": body}
body = await request.body()
request._receive = receive
return body
await proxy_get_body(request)
return await original_route_handler(request)
return custom_route_handler
然后在函数层内部读取的时候可以通过：
@app.post("/inapprsp")
async def inapprsp(req: Request, sdsada: AdminLogin):
print('inapprsp-接口内的：', id(req))
boay = await req._receive()
print("路由内多次读取：",boay)
boay = await req._receive()
print("路由内多次读取：",boay)
return 'inapprsp'
方案3：引入contextvars来存贮
首先contextvars本身处理其实就是上下文语境相关的模块，它是线程和协程安全的，可以很好处理。使用方式也简单，但是扩展实现的时候，显示的有点不是很优雅，感觉是突兀。这个的实现可以餐卡上一篇中调到关于链路追踪ID的实现方式。
自定义重写route，内部消费boay（这个后续有机会试一试）
总结
上面示例说明了几个问题点：
Fastapi【中间件内】内如果需要在【】请求开始】的时候读取body的是是无法处理的。 只能在中间件结束的时候读取 需要通过自定义类的方式来重写内部body消费问题，但是也只能在函数内部进行消费处理。（如果需要其其他中间件读取body的时候可以使用这种方式，如果不需要就完全没必要这么处理）
以上仅仅是个人结合自己的实际需求，做学习的实践笔记！如有笔误！欢迎批评指正！感谢各位大佬！
结尾
END
简书：https://www.jianshu.com/u/d6960089b087
掘金：https://juejin.cn/user/2963939079225608
公众号：微信搜【程序员小钟同学】
小钟同学 | 文 【欢迎一起学习交流】| QQ：308711822
原文作者：程序员小钟同学；如有侵权，请联系管理员删除。
请登录后发表评论
登录
全部评论
最新文章
- 深入FastAPI异步编程-之并发编程（15）-定义全局的支持全局上下文变量和局部上下文的对象可以任意传递了！
- 深入FastAPI异步编程-之并发编程（13）-纠错-修改基于aiojobs库结合aiojobs的和FastAPI框架的示例
- 深入FastAPI异步编程-之并发编程（13）-基于aiojobs库实现类似Background Tasks的效果
- 深入FastAPI异步编程-之并发编程（12）-基于pyee异步事件驱动通知
- 深入FastAPI异步编程-之并发编程（11）-异步中的?同步原语
- 深入FastAPI异步编程-之并发编程（10）-?asyncio.wait 更精细化并发补充说明
- 深入FastAPI异步编程-之并发编程（9）-asyncio.as_completed的补充说明
- 深入FastAPI异步编程-之并发编程（8）-asyncio.gather的补充说明
- 深入FastAPI异步编程-之并发编程（7）番外篇~Windows系统下事件循环插曲
- 深入FastAPI异步编程-之并发编程（6）~事件循环
