'Depends' object has no attribute '...' [HELP NEEDED]

[ricardo-reis-1970]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

credentials_exception = HTTPException(
    status_code=status.HTTP_401_UNAUTHORIZED,
    detail="Could not validate credentials",
    headers={"WWW-Authenticate": "Bearer"},
)

def get_user(username: str):
    user = User.get_by_username(username) # CHANGE
    if user:
        return user

async def get_current_user(token: str = Depends(oauth2_scheme)):
    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 = get_user(username=token_data.username) # CHANGE
    if user is None:
        raise credentials_exception
    return user

async def check_role(role: str):
    current_user = await get_current_user()
    print(f"[check_role] current_user: {current_user}")
    roles = [role.name for role in current_user.roles]
    return role in roles

async def check_admin():
    if await check_role('admin'):
        return True
    else:
        raise credentials_exception

Description

I'm trying to use security dependencies throughout my API, in order to make sure that endpoints are only available to the appropriate roles or privileges. Please note that in the above example, I'm defining a wrapper called check_admin because Depends won't allow me to pass the parameter 'admin' directly to check_role (what a shame!).

I'm using the example from https://fastapi.tiangolo.com/tutorial/security/oauth2-jwt/ with very small adjustments, such as getting the user from a SQLAlchemy call: User.get_by_username(username). I also pulled the credentials_exception out to the global scope, in order to use it from different places.

I've been moving away from Depends, because it's working in a rather unexpected way: instead of calling whatever dependency and returning its result, it's returning a Depends object, which I don't know what to do with.

For instance, I started with:

async def check_role(role: str, current_user: UserSchema = Depends(get_current_user)):
    roles = [role.name for role in current_user.roles]
    return role in roles

but I got:

AttributeError: 'Depends' object has no attribute 'roles'

Every time I convert what used to be a dependency into an explicit call, such as:

async def check_role(role: str):
    current_user = await get_current_user()
    roles = [role.name for role in current_user.roles]
    return role in roles

all I achieve is to propagate the Depends exception to the next function, in this case:

AttributeError: 'Depends' object has no attribute 'rsplit'

because get_current_user got the token Depends(OAuth2PasswordBearer) (instead of an actual token) and then, in the line:

payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

it raises the above exception.

From the docs — Dependencies - First Steps > Simple and Powerful —, I was under the impression that I could use any level of dependencies, and that these were precisely designed to simplify the code, instead of mixing everything in the 'man finction', working a bit like decorators:

Questions

• am I messing up things with async?
• should I only use Depends on endpoint functions?
• should I update any version (FastAPI, Python itself)?

This is blocking my entire API so far. Googling this leads always to the very documentation whose examples I'm unsuccessfully replicating. My only option has been to leave it totally insecure. I work in a highly regulated market. I need help.

Operating System

Windows

Operating System Details

No response

FastAPI Version

0.66.0

Python Version

3.9.0

Additional Context

No response

[Acerinth]

Hey there 👋🏼

Hmm, do you have a code example of using your Depends and how exactly are you passing the argument inside? Because, it works a bit differently then normal functions. I have implemented similar thing in my app as well, and it is important how you pass the arguments. E.g.

@project_router.get("/{project_id}", response_model=ProjectRead, response_model_exclude_none=True, 
                      responses={status.HTTP_404_NOT_FOUND: {"model": shared_schemas.Message}})
def read_project_by_id(project_id: str, db: Session = Depends(get_db), 
                      user: shared_schemas.UserSecurity = Security(get_current_user, scopes=["MY_SCOPE"])):

    project = controller.get_project(db, project_id)
    if not project:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found.")
    return project

The Security works almost the same as Depends, and implementation of my get_current_user is like so:

def get_current_user(security_scopes: SecurityScopes, token: HTTPAuthorizationCredentials = Depends(auth_scheme)):

    # here I first have some code to check if user token is valid ...

    # if the token is ok and I have the user object, I check the permissions:
    if user:
        for scope in security_scopes.scopes:
            if scope not in user.roles:
                raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="You do not have right permissions.")
        return user
    else:
        # raise some error

I have quite specific auth implementation as well, but that doesn't matter. I used the idea of scopes from tutorial here: https://fastapi.tiangolo.com/advanced/security/oauth2-scopes/ , so maybe this approach would simplify your roles check as well.

[ricardo-reis-1970]

Hi Paula,

Thank you for your reply. Your case seems to be where I should be going. I was trying to start small, just testing the Depends with some role check, but definitely using the scopes is interesting. The thing is that it kind of feels like for some particular features one should really test specific permissions and not an entire scope.

In any case, the docs and examples sometimes are reported as broken, which was precisely the case with Security. I could not stress enough how crucial for adoption it is to have fully tested and working bits of code. I'm moving from Flask, Fast looks way better, but...

Here's an initial attempt, consisting of 2 files:

utilities/login.py

from datetime import datetime, timedelta
from typing import Optional

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext

from ..models.user import User
from ..schemas.user import TokenData, UserCreate

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "82b9db72636d7254ca42b98b0ab006253abe6f7189a52cd6bd2eee81baf3d620"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

credentials_exception = HTTPException(
    status_code=status.HTTP_401_UNAUTHORIZED,
    detail="Could not validate credentials",
    headers={"WWW-Authenticate": "Bearer"},
)

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def get_user(username: str):
    user = User.get_by_username(username)
    if user:
        return user


def authenticate_user(username: str, password: str):
    user = get_user(username)
    if not user:
        return False
    if not verify_password(password, user.password):
        return False
    return user


def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(token: str = Depends(oauth2_scheme)):
    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 = get_user(username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user: UserCreate = Depends(get_current_user)):
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

def get_user_roles_and_permissions(user):
    roles = [role.name for role in user.roles]
    permissions = list({permission.name for role in user.roles for permission in role.permissions})
    return roles, permissions

async def check_role(role: str):
    current_user = await get_current_user()
    roles = [role.name for role in current_user.roles]
    return role in roles

async def check_admin():
    print("Checking 'admin'...")
    try:
        if await check_role('admin'):
            print('OK!')
            return True
        else:
            print('UNAUTHORISED!')
    except Exception as e:
        print(f"Exception {e}")

Regarding the line

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

there is a:

routers/login.py

that implements the /login endpoint like the one in the documentation:

@router.post("/")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    roles, permissions = get_user_roles_and_permissions(user)
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(data={"sub": user.username, "scopes": permissions}, expires_delta=access_token_expires)
    return {"access_token": access_token, "token_type": "bearer", "roles": roles, "permissions": permissions}

routers/test.py

from fastapi import APIRouter
from fastapi.param_functions import Depends

from ..utilities.login import oauth2_scheme, check_admin, get_current_active_user

router = APIRouter()

@router.get("")
async def all(ok: bool = Depends(check_admin)):
    print("ok: {ok}")

Obviously, my main.py does something like:

from manager_backend.routers import login, test

app.include_router(login.router, prefix="/login", tags=["Login"])
app.include_router(test.router, prefix="/test", tags=["Test"])

Now, whenever I try http://localhost:5000/test, I get the page reply:

// 20210818170017
// http://localhost:5000/test

null

and the cmd where I'm running the server says:

INFO:     127.0.0.1:49672 - "GET /test HTTP/1.1" 200 OK
Checking 'admin'...

[get_current_user] got Depends(OAuth2PasswordBearer)
Exception 'Depends' object has no attribute 'rsplit'
ok: None

I'm getting lost here. Any thoughts?

[Acerinth]

Regarding First part: scopes - this of course you can use and define however you want. I am using completely custom roles for this, and every endpoint requires different roles of course. So I think there you should be completely flexible. It's just a very handy argument you can use with Security :)

About your Depends issue - TBH, I am not sure if you are allowed to do this:
current_user = await get_current_user()
because, they are not like a normal "function"/"method" - at least I've never seen this kind of usage in official Tutorials. What I've seen, you usually use them always as a parameter of another method (that's how you have nested/sub-dependencies). Sooo, your check_role should be something like:

async def check_role(role: str, current_user = Depends(get_current_user)):
    roles = [role.name for role in current_user.roles]
    return role in roles

EDIT: And also, bcos your check_admin is the one you're calling from the router, you need to adjust that one as well:

async def check_admin(role = Depends(check_role, role='admin')):   # here I'm not sure how to pass your role into dependency, maybe there is something in tutorials about that... but that's why here those Security Scopes come in handy
    print("Checking 'admin'...")
    try:
        if role:
            print('OK!')
            return True
        else:
            print('UNAUTHORISED!')
    except Exception as e:
        print(f"Exception {e}")

You always need to propagate your Dependencies top -> down kinda (from router down, right)

Also, double check your types / objects that are returned from Depends, I just put up here the concept xD

Unfortunately, I didn't work with async/await, so I'm not sure if that would mess up stuff as well.

In any case, If this doesn't work - that's why I showed you how I solved user / permission check... In your case, I think you should use it something like this (test.py):

@router.get("")
async def all(user: User = Security(get_current_user, scopes=["admin"])):
    print("ok: {user}")

And then you could sort of copy-paste my/tutorials part about scopes (in your case == roles) for the get_current_user:

async def get_current_user(roles: SecurityScopes, token: str = Depends(oauth2_scheme)):
    # your stuff here....
    
    if user is None:
        raise credentials_exception
        
    # here check for your user's roles, perhaps there is no need for separating into methods as it should be 2-3 lines
    # and of course raise 403 or whatever when user doesn't have enough roles
        
    return user

I think this approach would be way easier... 🤷🏼 and it's not a big change anyways, so maybe you should give it a shot.

P.S: Using Security stuff this way, you have nice support for it in the API docs as well :)

[ricardo-reis-1970]

Dear @Acerinth,

Хвала вам!

You make some interesting points. I too have started to suspect that the async might be cramping the whole thing's style. The fragments throughout the documentation have (some) asyncs, but the final examples are all normal.

Funny you suggest I do:

async def check_role(role: str, current_user = Depends(get_current_user)):
    roles = [role.name for role in current_user.roles]
    return role in roles

This was precisely what I had, before I started getting that weird Depends object issue. I was trying to avoid the error, but all I did was push it down the ladder. I'm going to try it again. Maybe now it works.
EDIT: it does not, regrettably.

Regarding the parameter

role = Depends(check_role, role='admin')

we all wish this was possible, but the docs — Dependencies - First Steps — say:

Although you use Depends in the parameters of your function the same way you use Body, Query, etc, Depends works a bit differently.

You only give Depends a single parameter.

This parameter must be something like a function.

And that function takes parameters in the same way that path operation functions do.

I'm crying over the first bold and scratching my head over the second. The whole point for the function check_admin was that Depends does not take parameters.

This is Security and there are alternatives (such as you pointed out), but my faith on Depends is dropping very fast. I'm failing to see what's so special about it — other than helping psychiatrists — that explicit calls to auxiliary functions won't do.

You've been too kind. Thank you!

[Acerinth]

Hahahahhahaha, yeah. Sooo, obviously no usage of Depends in this way. I think it was mostly meant to be like it was described in the Tutorial... Database access, some repetitive query parameters for endpoints, etc. But obviously not for something that needs to take another parameter. I'm not a big fan of it either, but does the job great for DB stuff at least.

I still think it's worth checking out SecurityScopes class and what can you pack inside... For my purposes it was exactly enough to specify different permissions for every request. That should be able to solve your permission issues also.

In any case - you're welcome and good luck :D

[dionisnote]

Здравствуйте :)

Hi, @ricardo-reis-1970
I'm new in FastAPI, but I pretty like its Depends api(except its typings with Any). I think for your purpose you can use higher order function to setup role checking function, something like that

async def check_role(role: str):
   def is_users_role(current_user = Depends(get_current_user)):
        roles = [role.name for role in current_user.roles]
        return role in roles
    return is_users_role

So, then u can use it in this way:

@router.get("")
async def all(ok: bool = Depends(check_role("admin"))):
    print("ok: {ok}")

[ricardo-reis-1970]

@dionisnote

Gosh! Get out of my head! Or don't.

It's just that I was precisely considering currying or some other kind of higher order, and you come and show me just that! I feel like I must help someone now... or as soon as I know enough.

I am considering in the end just going with the Security solution and Scopes, but still I'd like to see this all through, out of sheer stubbornness. I kind of suspect that that my whole issue lies in the fact that one probably cannot interrupt the Depends chain. If this is the case, this is yet another pearl missing from the documentation necklace. Your solution actually keeps it.

Yes, but does it work?

Yes... almost!

In fact, when you do Depends(check_role("admin")), you're not passing it a function, but rather a promise (if a bit of JavaScript speak is allowed here). Since check_role itself is async, it will not return immediately, causing an error. So, the tiny detail is to remove the async from check_role, like this:

def check_role(role: str):
    async def is_users_role(current_user = Depends(get_current_user)):
        print(f"[check_role] current_user: {current_user}")
        roles = [role.name for role in current_user.roles]
        return role in roles
    return is_users_role

Notice how I transferred the async to is_users_role, just for fun and good measure, as it works fine without. The rule here is just not having it in the higher order function, as that one is not part of the mechanism, but rather part of the definition of said mechanism.

I hope this helps you as much as you helped me.

Other two interesting places of my head you seem to be in touch with are the fact that I much favour Functional Programming and the fact that I can actually read your Zdravstvuitie!

Cpacibo bolshoi!

[claytonwinders]

I've been fighting the same issue today, adapting the OAuth2 demo code and some of the fullstack-fast-api-postgres code. It seems to me that the chaining of the Depends statements is what is important. Chaining the demo code, the cookiecutter postgres code and my own code together at first I ran into the same Exception 'Depends' object has no attribute 'rsplit'. In hopes that this is helpful to someone else, the relevant snippets looked like:

# app/services/user_service.py

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/v1/user/token")
.....
async def get_current_user(self, token: str = Depends(oauth2_scheme)) -> UserData:
    try:
        payload = jwt.decode(token=token, key=settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise INVALID_CREDENTIALS
    except JWTError:
        raise INVALID_CREDENTIALS
    user = await self.check_user(user=UserCheck(user_name=username))
    if not user.checked:
        raise INVALID_CREDENTIALS
    current_user = await self.get_user(user=UserCheck(user_name=user.user_name))
    return current_user
-----
# app/api/deps.py

async def current_user(session: AsyncSession = Depends(get_session)) -> UserData:
    user_service = UserService(session=session)
    current_user = await user_service.get_current_user()
    return current_user
-----
# app/api/v1/endpoints/user.py

@router.get("/me", response_model=UserData)
async def current_user(current_user: UserData = Depends(current_user)):
    return current_user

By moving the Depends(oauth2_scheme) from get_current_user() in my app/services/user_service.py to current_user() in my app/api/deps.py and then passing the resulting returned value as passing a normal parameter, it is working.

# app/services/user_service.py

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/v1/user/token")
.....
# pass token directly here
async def get_current_user(self, token: str) -> UserData:
    try:
        payload = jwt.decode(token=token, key=settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise INVALID_CREDENTIALS
    except JWTError:
        raise INVALID_CREDENTIALS
    user = await self.check_user(user=UserCheck(user_name=username))
    if not user.checked:
        raise INVALID_CREDENTIALS
    current_user = await self.get_user(user=UserCheck(user_name=user.user_name))
    return current_user
---
# app/api/deps.py
async def current_user(session: AsyncSession = Depends(get_session), token: str = Depends(oauth2_scheme)) -> UserData:
    user_service = UserService(session=session)
    current_user = await user_service.get_current_user(token=token)
    return current_user

[johnshumon]

Worth reading this thread. mainly the first answer and also what @TheWindyMan mentioned above in his solution.

I ran into this exact same problem. Initially I wasn't passing db sessions to the function which was querying in the db if the user exists. Kept complaining about 'Depends' object has no attribute 'query'. As soon as I passed db session it worked just fine. So bottom line is chaining is important when using Depends

[zhbzhbzhbz]

@johnshumon thx~ That page helps a lot.

You can't use Depends in your own functions, it has to be in FastAPI functions, mainly routes. You can, however, use Depends in your own functions when that function is also a dependency, so could can have a chain of functions
What I do is get the DB session from the route and then pass it down through every layer and function. I believe this is also better design.