https://fastapi.tiangolo.com/zh/tutorial/middleware/

---

向 FastAPI 应用添加中间件.

"中间件"是一个函数,它在每个请求被特定的路径操作处理之前,以及在每个响应返回之前工作.

• 它接收你的应用程序的每一个请求.

• 然后它可以对这个请求做一些事情或者执行任何需要的代码.

• 然后它将请求传递给应用程序的其他部分 (通过某种路径操作).

• 然后它获取应用程序生产的响应 (通过某种路径操作).

• 它可以对该响应做些什么或者执行任何需要的代码.

• 然后它返回这个 响应.

创建中间件

要创建中间件你可以在函数的顶部使用装饰器 @app.middleware("http").

中间件参数接收如下参数:

• request.

• 一个函数 call_next 它将接收 request 作为参数.

  • 这个函数将 request 传递给相应的 路径操作.

  • 然后它将返回由相应的路径操作生成的 response.

• 然后你可以在返回 response 前进一步修改它.

import time

from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

高级中间件

https://fastapi.tiangolo.com/zh/advanced/middleware/

准ASGI中间件类

https://www.starlette.io/middleware/#ratelimitmiddleware

“Starlette includes several middleware classes for adding behavior that is applied across your entire application. These are all implemented as standard ASGI middleware classes, and can be applied either to Starlette or to any other ASGI application.”

---

实践

自定义用户鉴权的中间件

自定义中间件

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
class AuthenticationMiddleware(BaseHTTPMiddleware):
    def __init__(self, app):
        super().__init__(app)

    async def dispatch(self, request: Request, call_next):
        rt = self.check_authentication(request)
        if rt.status_code != status.HTTP_200_OK:
            error_response = JSONResponse(dict(detail=rt.error_msg), status_code=rt.status_code)
            return error_response
        # 认证成功,返回响应
        response = await call_next(request)
        return response

    def check_authentication(self, request: Request) -> UserRT:
		pass

调用中间件

from pydantic.error_wrappers import ValidationError
from starlette.middleware.cors import CORSMiddleware

app = FastAPI(
    title=settings.PROJECT_NAME,
    openapi_url=f"{settings.API_V1_STR}/openapi.json",
)

# 添加错误处理, 统一处理未捕获的pydantic ValidationError
app.add_exception_handler(ValidationError, pydantic_validation_error_handler)

# 添加中间件到FastAPI应用
app.add_middleware(AuthenticationMiddleware)