python：Fastapi – Middleware(中间件)

简单唠一些

最近有些忙，时间不怎富裕，学习总结也落下了，今天总结了一些，咱们继续书接上文。

继上篇文章主要唠了安全性的一部分知识(后续再补一些知识)，今天主要聊下中间件。

中间件

中间件（英语：Middleware），又译中间件、中介层，是一类提供系统软件和应用软件之间连接、便于软件各部件之间的沟

通的软件，应用软件可以借助中间件在不同的技术架构之间共享信息与资源。

而在Fastapi中的中间件是一个函数，用于处理每一请求路径操作之前，以及每个响应返回之前工作。它接收你应用程序的每

一个请求。它可以对这些请求做一些事情或执行任何需要的代码。它将通过某种路径操作将请求传递给应用程序的其他部分。它

还可以通过某种路径操作获取应用程序生产的响应，可以对该响应做些什么或者执行任何需要的代码，然后返回该响应。

import timeimport uvicornfrom fastapi import statusfrom fastapi import Requestfrom fastapi import FastAPIfrom pydantic import BaseModelfrom typing import Optionalfrom fastapi import Responseapp = FastAPI()@app.middleware("http")async def add_time_header(request: Request, call_):    start_time = time.time()    if request.method == "POST":        print(request.url)    response: Response = await call_(request)    process_time = time.time() - start_time    response.headers["X-Process-Time"] = str(process_time)    response.status_code = status.HTTP_201_CREATED    return responseclass Item(BaseModel):    name: str    description: str = "a description message!"    cc: Optional[str] = "2"@app.post("/items/", response_model=Item)async def read_item(item: Item):    return itemif __name__ == '__main__':    uvicorn.run("middleware_main:app", debug=True)

    def middleware(self, middleware_type: str) -> typing.Callable:        # 源码        assert (            middleware_type == "http"        ), 'Currently only middleware("http") is supported. 目前只支持中间件(“http”)。'        def decorator(func: typing.Callable) -> typing.Callable:            self.add_middleware(BaseHTTPMiddleware, dispatch=func)            return func        return decorator

源码这里写的其实就是含局部函数的方法，先assert断言是不是http，断言正确就执行局部函数decorator，否则就抛出AssertionError: Currently only middleware(“http”) is supported.

    if request.method == "POST":        print(request.url)

call_底层实际调用的是BaseHTTPMiddleware类:

<coroutine object BaseHTTPMiddleware.__call__.<locals>.call_next at 0x0000015ADA9219A0>

BaseHTTPMiddleware类的部分源码：

class BaseHTTPMiddleware:    def __init__(self, app: ASGIApp, dispatch: DispatchFunction = None) -> None:        self.app = app        self.dispatch_func = self.dispatch if dispatch is None else dispatch    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:            app_exc: typing.Optional[Exception] = None            send_stream, recv_stream = anyio.create_memory_object_stream()

在 response 的前和后

在任何路径操作收到request前,可以添加要和请求一起运行的代码。当然也可以在响应生成但是返回之前添加代码.

可以添加自定义请求头 X-Process-Time 包含以秒为单位的接收请求和生成响应的时间:

    # 往返回的头信息中增加一个参数    response.headers["X-Process-Time"] = str(process_time)    # 重新被赋值状态码    response.status_code = status.HTTP_201_CREATED

运行结果-截图如下：

今天先聊到这里吧，以上总结或许能帮助到你，或许帮助不到你，但还是希望能帮助到你，如有疑问、歧义，直接私信留言会及时修正发布；非常期待你的一键 3 连【 点赞、收藏、分享 】哟，谢谢！

未完成，待续……

一直在努力，希望你也是！