'Depends' object has no attribute '...' [HELP NEEDED] #8791
-
First Check
Commit to Help
Example Codecredentials_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_exceptionDescriptionI'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 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: I've been moving away from 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 rolesbut I got: 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 rolesall I achieve is to propagate the because 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
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 SystemWindows Operating System DetailsNo response FastAPI Version0.66.0 Python Version3.9.0 Additional ContextNo response |
Beta Was this translation helpful? Give feedback.
Replies: 11 comments
-
|
Hey there 👋🏼 Hmm, do you have a code example of using your @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 projectThe 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 errorI 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. |
Beta Was this translation helpful? Give feedback.
-
|
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 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:
|
Beta Was this translation helpful? Give feedback.
-
|
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 About your async def check_role(role: str, current_user = Depends(get_current_user)):
roles = [role.name for role in current_user.roles]
return role in rolesEDIT: And also, bcos your 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 ( @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 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 userI 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 :) |
Beta Was this translation helpful? Give feedback.
-
|
Dear @Acerinth, Хвала вам! You make some interesting points. I too have started to suspect that the 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 rolesThis was precisely what I had, before I started getting that weird Regarding the parameter role = Depends(check_role, role='admin')we all wish this was possible, but the docs — Dependencies - First Steps — say:
I'm crying over the first bold and scratching my head over the second. The whole point for the function This is Security and there are alternatives (such as you pointed out), but my faith on You've been too kind. Thank you! |
Beta Was this translation helpful? Give feedback.
-
|
Hahahahhahaha, yeah. Sooo, obviously no usage of I still think it's worth checking out In any case - you're welcome and good luck :D |
Beta Was this translation helpful? Give feedback.
-
|
Здравствуйте :) Hi, @ricardo-reis-1970 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_roleSo, then u can use it in this way: @router.get("")
async def all(ok: bool = Depends(check_role("admin"))):
print("ok: {ok}") |
Beta Was this translation helpful? Give feedback.
-
|
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 Yes, but does it work?Yes... almost! In fact, when you do 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_roleNotice how I transferred the 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! |
Beta Was this translation helpful? Give feedback.
-
|
I've been fighting the same issue today, adapting the OAuth2 demo code and some of the # 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_userBy moving the # 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 |
Beta Was this translation helpful? Give feedback.
-
|
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 |
Beta Was this translation helpful? Give feedback.
-
|
@johnshumon thx~ That page helps a lot.
|
Beta Was this translation helpful? Give feedback.
-
|
If you have problems with dependencies like shown in this discussion, I recommend you go to related section of FastAPI Docs ad read it carefully. After reading that you will likely not have such problems :) |
Beta Was this translation helpful? Give feedback.
If you have problems with dependencies like shown in this discussion, I recommend you go to related section of FastAPI Docs ad read it carefully. After reading that you will likely not have such problems :)