FastAPI - 如何在中间件内使用依赖项？

Question

我正在构建一个浏览器游戏，其中每个用户都有 4 种类型的资源，并且每个用户根据其农场的级别生产更多资源。

我想做的是，每当给定用户登录时，我想在他们刷新页面或执行任何操作时重新计算其当前资源。

中间件似乎是实现我目标的正确工具，但我对当前架构（多个路由器）的实现有点困惑。在执行任何其他 API 调用之前调用函数来执行资源重新计算的最简洁方法是什么？

这是我到目前为止所尝试过的（示例中间件）：

app.py（无中间件）：

from fastapi import FastAPI, Depends, Request
from src.api.v1.village import village_router
from src.api.v1.auth import auth_router
from src.api.v1.admin import admin_router
from src.core.auth import get_current_user
from src.core.config import *

def create_app() -> FastAPI:
    root_app = FastAPI()
    root_app.include_router(
        auth_router,
        prefix="/api/v1",
        tags=["auth"],
    )
    root_app.include_router(
        admin_router,
        prefix="/api/v1",
        tags=["admin"],
        dependencies=[Depends(get_current_user)],
    )
    root_app.include_router(
        village_router,
        prefix="/api/v1",
        tags=["village"],
    )
 
    return root_app
 

然后我添加了一个

helloworld

中间件，并将

get_current_user

添加为依赖项，因为用户必须登录才能执行计算。

app.py（带有中间件）：

from fastapi import FastAPI, Depends, Request
from src.api.v1.village import village_router
from src.api.v1.auth import auth_router
from src.api.v1.admin import admin_router
from src.core.auth import get_current_user
from src.core.config import *
import time


def create_app() -> FastAPI:
    root_app = FastAPI()
    root_app.include_router(
        auth_router,
        prefix="/api/v1",
        tags=["auth"],
    )
    root_app.include_router(
        admin_router,
        prefix="/api/v1",
        tags=["admin"],
        dependencies=[Depends(get_current_user)],
    )
    root_app.include_router(
        village_router,
        prefix="/api/v1",
        tags=["village"],
    )

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

    return root_app

似乎依赖项被忽略，因为即使我没有登录，中间件也会被调用，而我的 protected_routes 则不是这种情况（如果我没有登录，我的路线上会收到 401 错误）。

async def get_current_user(
    session=Depends(get_db), token: str = Depends(oauth2_scheme)
) -> UserAuth:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[AUTH_TOKEN_ALGO])
        email: str = payload.get("email")
        user_id: str = payload.get("user_id")
        if email is None:
            raise ValueError("A very specific bad thing happened.")
        token_data = UserJWTToken(user_id=user_id, email=email)
    except jwt.PyJWTError:
        raise ValueError("A very specific bad thing happened.")
    user = get_user_by_email(session, token_data.email)
    if user is None:
        raise ValueError("A very specific bad thing happened.")
    return user

Answer 1

您可以使用全局依赖项。这是一个在这种情况下可能对您有所帮助的示例

from fastapi import Depends, FastAPI, Request


def get_db_session():
    print("Calling 'get_db_session(...)'")
    return "Some Value"


def get_current_user(session=Depends(get_db_session)):
    print("Calling 'get_current_user(...)'")
    return session


def recalculate_resources(request: Request, current_user=Depends(get_current_user)):
    print("calling 'recalculate_resources(...)'")
    request.state.foo = current_user


app = FastAPI(dependencies=[Depends(recalculate_resources)])


@app.get("/")
async def root(request: Request):
    return {"foo_from_dependency": request.state.foo}