'FastAPI - dependency inside Middleware?

I am building a browser game where every user has 4 types of ressources and each users produce more ressources based on the level of their farms.

What I am trying to do, is whenever a given user is logged in, I want to recalculate it's current ressources whenever he is refreshing the page or performing any action.

Seems the middleware is the right tool for my need but I am a bit confused on the implementation with my current architecture (multiple routers). What would be the cleanest way to call a function to perform ressources recalculation before performing any other API calls?

This is what I have tried so far (example middleware):

app.py (without middleware):

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
 

I then added an helloworld middleware and add the get_current_user as a dependency because a user must be logged_in to perform the calculations.

app.py (with middleware):

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

Seems the dependency is ignored because the middleware is called even when I am not logged in, which is not the case for my protected_routes (I am getting a 401 error on my routes if I a not logged in).

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


Sources

This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.

Source: Stack Overflow

Solution Source