Search code examples
pythonoauth-2.0fastapi

make Depends optional in fastapi python


I have an API which gives images to the user and nonusers. Images can be public or private.

My code

@router.get("/{id}")
def get_resource(id: str, current_user: User = Depends(get_current_user)):
  return return_resource(id, current_user)

This code enforces authorization strictly. I want if user is not logged in then it should put None in current_user so that I can allow access to public images and restrict private.

Other codes

get_current_user

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")

async def get_current_user(required: bool = True, token: str = Depends(oauth2_scheme)):
  credentials_exception = HTTPException(
    status_code=status.HTTP_401_UNAUTHORIZED,
    detail="Could not validate credentials",
    headers={"WWW-Authenticate": "Bearer"},
  )

  if not required and not token:
    return None

  return verify_token(token, credentials_exception)

I want to send parameter like required to get_current_user

verify_token

def verify_token(token: str, credentials_exception):
  try:
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    email: str = payload.get("email")
    pk: str = payload.get("pk")
    if email is None:
        raise credentials_exception
    token_data = TokenData(
        email=email,
        pk = pk
    )
  except JWTError:
      raise credentials_exception
  return token_data

Solution

  • If I understand you correctly, you can use a wrapper function to pass a parameter to a nested function. Like so:

    def get_current_user(required: bool = True):
        async def _get_user(token: str = Depends(oauth2_scheme)):
            credentials_exception = HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Could not validate credentials",
                headers={"WWW-Authenticate": "Bearer"},
            )
    
            if not required and not token:
                return None
    
            return verify_token(token, credentials_exception)
    
        return _get_user
    
    
    
    @router.get("/{id}")
    def get_resource(id: str, current_user: User = Depends(get_current_user(False))):
      return return_resource(id, current_user)