How to get current active user in middleware FastAPI python

Question

I have developed an auth on FastAPI Python using the doc here https://fastapi.tiangolo.com/tutorial/security/oauth2-jwt/ (I use exactly the code shown, nothing special to mention, auth works like a charm);

Now, I need to log things in a database at each call to each route: time spent for the request, method for current req, etc. - and username (the one that used the path).

For that I have developed an http middleware that looks like this:

@app.middleware("http")
async def log_things(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)

    path = request.url.path
    method = request["method"]
    params = request.path_params
    host = request.client.host + ":" + str(request.client.port)
    process_time = time.time() - start_time
    user = await security_management.get_current_active_user()  
    user = "dev" # MOCK todo how to get user

    db.write_log(path, method, params, host, process_time, user)

    response.headers["X-Process-Time"] = str(process_time)
    return response

Problem: I am unable to retrieve usernames. The content I retrieve is encapsulated in a "Depend" object that I am unable to resolve.

I've tried few things:

• using await on Depend,
• using current_user: User = Depends(get_current_user) as a parameter for my middleware function,

I'm kinda stuck, unable to extract valuable data from this Depend object, even if it works very well in the security part with these two functions below (that I tried to reproduce as well in terms of parameters...)

async def get_current_user(
        token: str = Depends(oauth2_scheme),
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = UserManager.get_user(real_user_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user # I can read current_user well...

Any thoughts? Thanks :)

Answer 1

I had exactly same issue in my application and came across a workaround/solution.

    @app.middleware("http")
    async def request_middleware(request, call_next):
    
        # some operation
    
        if request.headers.get('Authorization'):
            HttpRequestUtil.set_current_user_context(request=request)
    
        return await call_next(request)


    class HttpRequestUtil:
    
        @staticmethod
        def get_bearer_token(request: Request):
            auth_token = request.headers.get('Authorization')
            if 'Bearer' in auth_token:
                bearer_token: str = auth_token.split('Bearer')[1].strip()
                return bearer_token
    
        @staticmethod
        def set_current_user_context(request: Request):
            jwt_token=HttpRequestUtil.get_bearer_token(request)
            
            # if you want jwt claims, you can do below operation
            jwt_claims = jwt.get_unverified_claims(jwt_token)
            # ...
            # ...
            # Other Operation
            # ...