Reputation: 227
I'm struggling to find a decent implementation to return a validated user with class implementation. Previously it worked with simple functions, but I want to refactor this piece. So I added a class and trying to make it work. The problem is it doesn't seem to create a new instance of a class. If I'm trying to do Depends(UserAuthService().test)
I'm getting an error
{
"message": "Bad request.",
"details": "'TokenService' object is not callable"
}
router.py
router = APIRouter()
@router.get("/verify")
def verify_user(user: User = Depends(UserAuthService().get_current_valid_user)):
return user
user_auth_service.py
class UserAuthService:
def __init__(self):
self.user_repository = UserRepository()
self.token_service = TokenService()
def get_current_user(self, token: str = Depends(OAUTH2_SCHEME)):
credentials_exception = HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Could not validate the token"
)
user_does_not_exist_exception = HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="User does not exist"
)
try:
payload = self.token_service.decode_access_token(token)
sub: int = payload.get("sub")
if sub is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = self.user_repository.get_user(sub)
if user is None:
raise user_does_not_exist_exception
return User(**user)
def get_current_valid_user(self, user: User = Depends(get_current_user)):
if user.disabled:
raise HTTPException(status_code=400, detail="Inactive user")
return user
def test(self):
return self.token_service('123456')
P.S. as you can see there's Depends inside class methods, but that's from previous functional implementation.
Upvotes: 2
Views: 4189
Reputation: 227
After fixing classes the final code looks like this
router.py
@router.get("/verify")
def verify_user(user: User = Depends(UserValidator())):
return user
user_auth_service.py and user_validator.py
class UserAuthService:
def __init__(self):
self.user_repository = UserRepository()
self.token_service = TokenService()
self.oauth2_scheme = OAuth2BearerCookie()
async def get_current_user(self, request: Request):
token = await self.oauth2_scheme(request)
credentials_exception = HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Could not validate the token"
)
user_does_not_exist_exception = HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="User does not exist"
)
try:
payload = self.token_service.decode_access_token(token)
sub: int = payload.get("sub")
if sub is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = self.user_repository.get_user(sub)
if user is None:
raise user_does_not_exist_exception
return User(**user)
async def get_current_valid_user(self, request: Request):
user = await self.get_current_user(request)
if user.disabled:
raise HTTPException(status_code=400, detail="Inactive user")
return user
class UserValidator:
async def __call__(self, request: Request):
user_service = UserAuthService()
return await user_service.get_current_valid_user(request)
Upvotes: 1
Reputation: 9364
class UserAuthService:
def __init__(self):
...
self.token_service = TokenService()
...
def test(self):
return self.token_service('123456')
class A():
def __init__(self, param = None):
self.param = param
def print_something(self, something = None):
print('Init param is', self.param)
print('Something is', something)
class B():
def __init__(self, param = None):
self.a_instance = A(param)
def test(self, something = None):
self.a_instance.print_something(something) # run a_instance.print_something() instead of a_instance()
b = B('A obj')
b.test('test')
Init param is A obj
Something is test
class ACallable():
def __init__(self, param = None):
self.param = param
def __call__(self, something = None):
print('Init param is', self.param)
print('Something is', something)
class BCalling():
def __init__(self, param = None):
self.a_instance = ACallable(param)
def test(self, something = None):
self.a_instance(something) # __call__ method will be called
b = BCalling('A obj')
b.test('test')
Init param is A obj
Something is test
class A():
def __init__(self, param = None):
print('Init param is', param)
class B():
def __init__(self, param = None):
self.a_instance = A # here is A without brackets
def test(self, param = None):
self.a_instance(param) # here is a_instance (== A) with brackets
b = B()
b.test('<- called from constructor')
Init param is <- called from constructor
Upvotes: 4