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

Question

我已经阅读了 FastAPI 的关于中间件的文档（特别是 middleware tutorial、CORS middleware section 和 advanced middleware 指南），但找不到如何编写中间件类的具体示例，您可以使用添加add_middleware 函数（与使用装饰器添加的基本中间件函数相反）在此站点上也没有。

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

所以我的问题是：你是怎么做到的？

Answer 1

由于 FastAPI 实际上是Starlette，您可以使用BaseHTTPMiddleware，它允许您实现中间件类（您可能还想看看this post）。下面给出了相同方法的两种变体，其中add_middleware() 函数用于添加中间件类。

选项 1

middleware.py

from fastapi import Request

class MyMiddleware:
    def __init__(
            self,
            some_attribute: str,
    ):
        self.some_attribute = some_attribute

    async def __call__(self, request: Request, call_next):
        # do something with the request object
        content_type = request.headers.get('Content-Type')
        print(content_type)
        
        # process the request and get the response    
        response = await call_next(request)
        
        return response

app.py

from fastapi import FastAPI
from middleware import MyMiddleware
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()
my_middleware = MyMiddleware(some_attribute="some_attribute_here_if_needed")
app.add_middleware(BaseHTTPMiddleware, dispatch=my_middleware)

选项 2

middleware.py

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

class MyMiddleware(BaseHTTPMiddleware):
    def __init__(
            self,
            app,
            some_attribute: str,
    ):
        super().__init__(app)
        self.some_attribute = some_attribute

    async def dispatch(self, request: Request, call_next):
        # do something with the request object, for example
        content_type = request.headers.get('Content-Type')
        print(content_type)
        
        # process the request and get the response    
        response = await call_next(request)
        
        return response

app.py

from fastapi import FastAPI
from middleware import MyMiddleware

app = FastAPI()
app.add_middleware(MyMiddleware, some_attribute="some_attribute_here_if_needed")

Answer 2

@Error - Syntactical Remorse 引发的 BaseHTTPMiddleware bug 的潜在解决方法是使用 partial 并使用函数式方法来定义中间件：

中间件.py

​​>

from typing import Any, Callable, Coroutine


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

app.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)