如何编写自定义 FastAPI 中间件类

Question

我已经阅读了 FastAPI 关于中间件的文档（特别是中间件教程、CORS 中间件部分和高级中间件指南），但找不到如何编写可以使用该add_middleware函数添加的中间件类的具体示例（相反到使用装饰器添加的基本中间件功能），也不在此站点上。

我更喜欢使用add_middleware基于应用程序的装饰器的原因是，我想在共享库中编写一个中间件，该库将由多个不同的项目使用，因此我无法将其绑定到特定实例FastAPI。

所以我的问题是：你是如何做到的？

Answer 1

由于 FastAPI 实际上是Starlette，您可以使用BaseHTTPMiddleware它来实现中间件类（您可能还想看看这篇文章）。下面给出了相同方法的两个变体，其中该add_middleware()函数用于添加中间件类。请注意，目前无法与\xe2\x80\x94 一起使用BackgroundTasks（如果这是您的任务的要求），请检查#1438和#1640以了解更多详细信息。替代方案可以在这个答案和这个答案中找到。BaseHTTPMiddleware

\n

选项1

\n

中间件.py

\n

from fastapi import Request\n\nclass MyMiddleware:\n    def __init__(\n            self,\n            some_attribute: str,\n    ):\n        self.some_attribute = some_attribute\n\n    async def __call__(self, request: Request, call_next):\n        # do something with the request object\n        content_type = request.headers.get(\'Content-Type\')\n        print(content_type)\n        \n        # process the request and get the response    \n        response = await call_next(request)\n        \n        return response\n

\n

应用程序.py

\n

from fastapi import FastAPI\nfrom middleware import MyMiddleware\nfrom starlette.middleware.base import BaseHTTPMiddleware\n\napp = FastAPI()\nmy_middleware = MyMiddleware(some_attribute="some_attribute_here_if_needed")\napp.add_middleware(BaseHTTPMiddleware, dispatch=my_middleware)\n

\n

选项2

\n

中间件.py

\n

from fastapi import Request\nfrom starlette.middleware.base import BaseHTTPMiddleware\n\nclass MyMiddleware(BaseHTTPMiddleware):\n    def __init__(\n            self,\n            app,\n            some_attribute: str,\n    ):\n        super().__init__(app)\n        self.some_attribute = some_attribute\n\n    async def dispatch(self, request: Request, call_next):\n        # do something with the request object, for example\n        content_type = request.headers.get(\'Content-Type\')\n        print(content_type)\n        \n        # process the request and get the response    \n        response = await call_next(request)\n        \n        return response\n

\n

应用程序.py

\n

from fastapi import FastAPI\nfrom middleware import MyMiddleware\n\napp = FastAPI()\napp.add_middleware(MyMiddleware, some_attribute="some_attribute_here_if_needed")\n

\n

Answer 2

@Error - Syntropical Remorse 提出的BaseHTTPMiddleware bug的一个潜在解决方法，似乎至少对我有用，是使用部分方法并使用函数方法来定义中间件：

中间件.py

from typing import Any, Callable, Coroutine
from fastapi import Response


async def my_middleware(request: Request, call_next: Callable, some_attribute: Any) -> Response:
    request.state.attr = some_attribute  # Do what you need with your attribute
    return await call_next(request)

应用程序.py

from functools import partial
from fastapi import FastAPI
from middleware import my_middleware


app = FastAPI()

my_custom_middleware: partial[Coroutine[Any, Any, Any]] = partial(my_middleware, some_attribute="my-app")

app.middleware("http")(my_custom_middlware)