前言

关于FASTapi中body消息问题再研究，其实一直存在误区：能不能再中间件里面读取body，这里经过研究一次源码后，其实发现，之前分析不够透彻，于是乎再来补充一次！！

这次需要说明的是其实中间件也可以读取消息体！！！这次需要说明的是其实中间件也可以读取消息体！！！这次需要说明的是其实中间件也可以读取消息体！！！

如下示例diamante是无法消费示例：

from fastapi import FastAPI, Header
from fastapi.responses import HTMLResponse, StreamingResponse, FileResponse

import typing


from starlette.requests import Request
from starlette.responses import Response, StreamingResponse, PlainTextResponse
from starlette.types import ASGIApp, Receive, Scope, Send
from fastapi import Depends

RequestResponseEndpoint = typing.Callable[[Request], typing.Awaitable[Response]]
DispatchFunction = typing.Callable[
    [Request, RequestResponseEndpoint], typing.Awaitable[Response]
]
from starlette.middleware.base import BaseHTTPMiddleware


class LogerMiddleware1(BaseHTTPMiddleware):

    async def dispatch(self, request: Request, call_next):
        _body = await request.body()

        response = await call_next(request)

        return response


class ACORSMiddleware:
    def __init__(
            self,
            app: ASGIApp
    ) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":  # pragma: no cover
            await self.app(scope, receive, send)
            return
        res = Request(scope, receive)
        _json = await res.json()
        print("_json", _json)
        await self.app(scope, receive, send)


app = FastAPI()

app.add_middleware(ACORSMiddleware)
# app.add_middleware(LogerMiddleware1)

# app.add_middleware(LogerMiddleware1)

@app.get("/")
async def main(res: Request):
    # 迭代的方式返回流数据
    # buas = await res.json()
    # print("asssssssssssssss", buas)
    return PlainTextResponse('fanhui')


from pydantic import BaseModel


class AdminLogin(BaseModel):
    """
    管理员登陆 提交的参数信息
    """
    password: str = None
    username: str = None


@app.get("/on")
async def main(res: Request, sdsada: AdminLogin = Depends()):
    # 迭代的方式返回流数据
    # buas = await res.json()
    # print("asssssssssssssss", buas)
    return PlainTextResponse('fanhui')


if __name__ == '__main__':
    import uvicorn

    uvicorn.run('zhogjian:app')

上面示例中：

如果同时使用两个中间且两个中间件中都想要读取_body = await request.body() 则无法进行
如果只是单纯一个中间件中进行读取，是可以进行读取，但是只是，如果路由中也涉及到读取，如：

@app.get("/on")
async def main(res: Request, sdsada: AdminLogin ):
    # 迭代的方式返回流数据
    # buas = await res.json()
    # print("asssssssssssssss", buas)
    return PlainTextResponse('fanhui')

注意不是：

sdsada: AdminLogin = Depends() 这种是查询参数的解析

而下面对的这种才是涉及body解析：

 sdsada: AdminLogin ):

所以当你中间件读取了，然后又有接口涉及到继续进行参数解析的时候，那么测试中间件见消费request.body()则不成立了！

所以双重request.body()得到解析读取还是会有问题。但但但！！！！！！！！！！！！！！！ ----又有一种新的情况，假如我只是使用一个中间件和一个路由去消费，但是这个中间件我换一种方式注册：

原来的注册方式：



async def set_body(request):
    receive_ = await request._receive()
    async def receive():
        return receive_
    request._receive = receive


class ACORSMiddleware:
    def __init__(
            self,
            app: ASGIApp
    ) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":  # pragma: no cover
            await self.app(scope, receive, send)
            return
        request = Request(scope, receive)
        await set_body(request)
        re = await request.json()
        print("中间件消费", re)
        # _json = await res.json()
        print("_json", re)
        await self.app(scope, receive, send)

现在的两一种方式：


async def set_body(request):
    receive_ = await request._receive()
    async def receive():
        return receive_
    request._receive = receive

app = FastAPI()
@app.middleware("http")
async def token_update(request, call_next):
    """验证token是否过期，未过期则刷新token"""
    # 过滤无需相关操作的接口请求地址
    await set_body(request)
    re = await request.json()
    print("中间件消费", re)
    response = await call_next(request)
    return response
    
或者：
class LogerMiddleware1(BaseHTTPMiddleware):

    async def set_body(self,request):
    receive_ = await request._receive()
    async def receive():
        return receive_
    request._receive = receive

    async def dispatch(self, request: Request, call_next):
        await set_body(request)
        re = await self.request.json()
        print("中间件消费", re)
        response = await call_next(request)

        return response
app.add_middleware(LogerMiddleware1)

更换之后，那么此时路由和中间件中都可以同时进行消费！！！！！！！对比自定义和基于基础中间件的代码看：

基础中间件消费rquest

自定义而定：

class ACORSMiddleware:
    def __init__(
            self,
            app: ASGIApp
    ) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":  # pragma: no cover
            await self.app(scope, receive, send)
            return
        request = Request(scope, receive)
        await set_body(request)
        re = await request.json()
        print("中间件消费", re)
        # _json = await res.json()
        print("_json", re)
        await self.app(scope, receive, send)

完整单独一个中间件+路由双重可以消费的示例：

from fastapi import FastAPI, Header
from starlette.requests import Request
from starlette.responses import Response, StreamingResponse, PlainTextResponse
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()

from pydantic import BaseModel


class AdminLogin(BaseModel):
    """
    管理员登陆 提交的参数信息
    """
    password: str = None
    username: str = None


class LogerMiddleware1(BaseHTTPMiddleware):

   async def set_body(self, request):
        # 定义一个携程，返回是一个
        async def receive():
            return receive_
        receive_ = await request._receive()
        request._receive = receive

    async def dispatch(self, request: Request, call_next):
        await self.set_body(request)
        re = await request.json()
        print("中间件消费", re)
        response = await call_next(request)
        return response


app.add_middleware(LogerMiddleware1)


@app.get("/on")
async def main(res: Request, sdsada: AdminLogin):
    # 迭代的方式返回流数据
    buas = await res.json()
    print("路由中在消费---》", buas)
    return PlainTextResponse('fanhui')


if __name__ == '__main__':
    import uvicorn
    uvicorn.run('zhogjian:app')

关键核心的代码其实就是：

 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        async def call_next(request: Request) -> Response:
            ·····
            return response
        # 实现中间件消费这里也很关键
        async with anyio.create_task_group() as task_group:
            request = Request(scope, receive=receive)
            response = await self.dispatch_func(request, call_next)
            await response(scope, receive, send)
            task_group.cancel_scope.cancel()

另外就是这里的协程对象临时变量的重新赋值：

   async def set_body(self, request):
        # 定义一个携程对象，返回是一个
        async def receive():
            return receive_
        receive_ = await request._receive()
        # 读取返回的是{'type': 'http.request', 'body': b'{\r\n    "saaaaaaaaas":322222222222\r\n}', 'more_body': False}
        # 把返回的直接赋值给_receive 这样就避免了消费后就无法消费，相当于我消费过把结果存贮，在赋值给原来的_receive，让下一次在进行在消费的时候再读取
        request._receive = receive

以上仅仅是个人结合自己的实际需求，做学习的实践笔记！如有笔误！欢迎批评指正！感谢各位大佬！

结尾

END

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

公众号：微信搜【程序员小钟同学】

小钟同学 | 文【欢迎一起学习交流】| QQ：308711822

ITPub博客

fastapi框架源码分析系列-starlette框架之Debug方式学源码（5）fastapi中间件消费body消息体再次消费

前言

基础中间件消费rquest

自定义而定：

结尾

END

最新文章