我正在构建一个浏览器游戏，其中每个用户都有 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

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}