CODEWITHSUNDEEP
pythonfastapimiddleware
Pierre-Alexandre
Pierre-Alexandre

Reputation: 755

FastAPI - How to use dependencies inside a Middleware?

I am building a browser game where every user has 4 types of resources and each user produces more resources 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 its current resources whenever they are refreshing the page or performing any action.

It seems that the middleware is the right tool for my goal, 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 resources recalculation before performing any other API call?

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 a helloworld middleware and added the get_current_user as a dependency, because a user must be logged in, in order 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

Upvotes: 7

Views: 7524

Answers (1)

JPG
JPG

Reputation: 88429

You can make use of the Global Dependencies. Here is one example that may help you in this situation

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}

Upvotes: 5

Related Questions