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

Question

我已经阅读了 FastAPI 有关中间件的文档（具体来说，中间件教程、CORS 中间件部分和高级中间件指南），但找不到如何编写中间件类的具体示例，您可以使用

add_middleware

函数添加（与使用装饰器添加的基本中间件函数相反），无论是在本网站上还是在该网站上。

我更喜欢使用

add_middleware

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

FastAPI 

实例。

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

Answer 1

由于 FastAPI 实际上是下面的 Starlette，因此您可以使用

BaseHTTPMiddleware

来实现中间件类（您可能还想看看 this post）。下面给出了相同方法的两种变体，其中

add_middleware()

函数用于添加中间件类。请注意，目前无法将

BackgroundTasks

（如果这是您的任务的要求）与

BaseHTTPMiddleware

一起使用 — 请检查 #1438 和 #1640 了解更多详细信息。替代方案可以在这个答案和这个答案中找到。

选项1

中间件.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

中间件.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 - Syntropical Remorse 提出的 BaseHTTPMiddleware bug 的一个潜在解决方法，至少对我来说似乎有效，是使用 partial 并使用函数式方法来定义中间件：

中间件.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)