You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

FastAPI中如何让自定义中间件优先于AuthenticationMiddleware执行?

问题:自定义中间件无法在AuthenticationMiddleware前执行

我在向挂载了Starlette应用的FastAPI服务的Cookie中传递id_token,希望自定义中间件能在AuthenticationMiddleware之前执行以完成token验证,但目前AuthenticationMiddleware总是先运行。

代码示例

import uvicorn
from fastapi import FastAPI
from starlette.applications import Starlette
from starlette.authentication import (
    AuthCredentials,
    AuthenticationBackend,
    BaseUser,
)
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
from starlette.routing import Route


class AuthenticatedUser(BaseUser):
    def __init__(self, display_name: str) -> None:
        self._display_name = display_name

    @property
    def is_authenticated(self) -> bool:
        return True


class AuthBackend(AuthenticationBackend):
    async def authenticate(self, conn):
        print("Authenticate method fired")
        return AuthCredentials(["authenticated"]), AuthenticatedUser(
            display_name="Test User"
        )


class CustomHeaderMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        print("custom middleware fired")
        return response


middleware = [
    Middleware(CustomHeaderMiddleware),
    Middleware(AuthenticationMiddleware, backend=AuthBackend()),
]


async def starlette_homepage(request):
    return JSONResponse({"message": "Hello from Starlette!"})


starlette_app = Starlette(
    debug=True,
    routes=[
        Route("/", starlette_homepage),
    ],
)

app = FastAPI(middleware=middleware)


@app.get("/")
async def fastapi_homepage():
    return {"message": "Hello from FastAPI!"}


app.mount("/starlette", starlette_app)

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

运行日志

INFO:     127.0.0.1:51118 - "GET / HTTP/1.1" 200 OK
Authenticate method fired
custom middleware fired
INFO:     127.0.0.1:51129 - "GET /starlette HTTP/1.1" 307 Temporary Redirect
Authenticate method fired
custom middleware fired
INFO:     127.0.0.1:51129 - "GET /starlette/ HTTP/1.1" 200 OK

请问如何确保自定义中间件优先执行,以便在AuthenticationMiddleware创建AuthenticatedUser前完成token验证?


解决方法

问题核心是BaseHTTPMiddleware的执行机制:它属于请求后中间件,代码中await call_next(request)会先触发后续中间件和路由逻辑,之后才会运行print("custom middleware fired")这类代码,所以看起来AuthenticationMiddleware先执行了。

要让自定义中间件在AuthenticationMiddleware之前处理请求,需要将其改成标准ASGI中间件,而非继承BaseHTTPMiddleware。

修改后的代码

import uvicorn
from fastapi import FastAPI
from starlette.applications import Starlette
from starlette.authentication import (
    AuthCredentials,
    AuthenticationBackend,
    BaseUser,
)
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.responses import JSONResponse
from starlette.routing import Route


class AuthenticatedUser(BaseUser):
    def __init__(self, display_name: str) -> None:
        self._display_name = display_name

    @property
    def is_authenticated(self) -> bool:
        return True


class AuthBackend(AuthenticationBackend):
    async def authenticate(self, conn):
        print("Authenticate method fired")
        return AuthCredentials(["authenticated"]), AuthenticatedUser(
            display_name="Test User"
        )


# 改为标准ASGI中间件
class CustomTokenMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        # 在此处执行token验证逻辑,比如从Cookie提取id_token
        if scope["type"] == "http":
            print("custom middleware fired (token validation done)")
            # 示例:获取Cookie中的id_token
            # cookies = scope.get("cookies", {})
            # id_token = cookies.get("id_token")
            # 执行解码、验证逻辑,可将用户信息存入scope供后续使用
            # scope["user_info"] = decoded_user_data

        # 调用后续中间件/应用逻辑
        await self.app(scope, receive, send)


middleware = [
    Middleware(CustomTokenMiddleware),
    Middleware(AuthenticationMiddleware, backend=AuthBackend()),
]


async def starlette_homepage(request):
    return JSONResponse({"message": "Hello from Starlette!"})


starlette_app = Starlette(
    debug=True,
    routes=[
        Route("/", starlette_homepage),
    ],
)

app = FastAPI(middleware=middleware)


@app.get("/")
async def fastapi_homepage():
    return {"message": "Hello from FastAPI!"}


app.mount("/starlette", starlette_app)

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

关键说明

  1. 标准ASGI中间件结构:通过__init__接收应用实例,__call__方法处理请求,在await self.app(scope, receive, send)之前的代码会优先于后续中间件执行。
  2. 验证逻辑位置:把token提取、验证逻辑放在await self.app(...)之前,确保在AuthenticationMiddleware的authenticate方法触发前完成验证,还能将验证后的用户信息存入scope,供后续中间件或路由调用。
  3. 中间件顺序:保持middleware列表中CustomTokenMiddleware在AuthenticationMiddleware之前,ASGI中间件按列表顺序从外到内包裹应用,先定义的先执行请求处理逻辑。

修改后运行日志会变为:

custom middleware fired (token validation done)
Authenticate method fired
INFO:     127.0.0.1:xxxx - "GET / HTTP/1.1" 200 OK

内容的提问来源于stack exchange,提问作者PGHE

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.18 11:17:06