refactoring some code by removing unnecessary checks and moving tests block to root /
This commit is contained in:
+21
-13
@@ -22,6 +22,7 @@ class CurrentUser:
|
||||
def _check(self, form_data_email:str, form_data_password:str,):
|
||||
'''check user by email'''
|
||||
user=self.crud_db_actions.get_user_by_email(form_data_email)
|
||||
|
||||
if user is None:
|
||||
raise self.error.credentials_error(detail="Wrong credentials")
|
||||
|
||||
@@ -36,20 +37,21 @@ class CurrentUser:
|
||||
def get_current_user(self, token:str)->UserOut:
|
||||
|
||||
payload=self.jwt_service.jwt_decode(token)
|
||||
sub=payload.get("sub")
|
||||
|
||||
if (sub:=payload.get("sub")) is None:
|
||||
raise self.error.credentials_error(detail="Jwt token is incorrect")
|
||||
|
||||
try:
|
||||
sub=UUID(sub)
|
||||
except (ValueError, TypeError):
|
||||
raise self.error.credentials_error(detail="Jwt token is incorrect")
|
||||
except (ValueError, TypeError) as e:
|
||||
raise self.error.credentials_error(detail="Jwt token is incorrect") from e
|
||||
|
||||
|
||||
user=self.crud_db_actions.get_user_by_id(sub)
|
||||
if user is None:
|
||||
raise self.error.not_found_error(detail="User with this email address not found")
|
||||
|
||||
if user.status is False:
|
||||
raise self.error.credentials_error(detail="This user is deactivated")
|
||||
|
||||
return UserOut.model_validate(user)
|
||||
|
||||
|
||||
@@ -61,12 +63,13 @@ class CurrentUser:
|
||||
|
||||
|
||||
def create_refresh_token(self,user_id:UUID, request:Request)->str:
|
||||
|
||||
token, jti=self.jwt_service.create_refresh_token({"sub":str(user_id)})
|
||||
|
||||
try:
|
||||
jti=UUID(jti)
|
||||
except (ValueError, TypeError):
|
||||
raise self.error.credentials_error(detail="Jwt token is incorrect")
|
||||
except (ValueError, TypeError) as e:
|
||||
raise self.error.credentials_error(detail="Jwt token is incorrect") from e
|
||||
|
||||
'''create new refresh token if all the checks are successful'''
|
||||
token_record=RefreshTokensCreate(
|
||||
@@ -83,16 +86,19 @@ class CurrentUser:
|
||||
|
||||
|
||||
def refresh_token(self, refresh_token:str, request:Request)->tuple[str, str]:
|
||||
|
||||
'''decode old refresh token'''
|
||||
old_refresh_token=self.jwt_service.jwt_decode(refresh_token)
|
||||
if (sub:=old_refresh_token.get("sub")) is None or (old_jti:=old_refresh_token.get("jti")) is None:
|
||||
sub=old_refresh_token.get("sub")
|
||||
|
||||
if (old_jti:=old_refresh_token.get("jti")) is None:
|
||||
raise self.error.credentials_error(detail="Jwt token is incorrect")
|
||||
|
||||
try:
|
||||
old_jti=UUID(old_jti)
|
||||
sub=UUID(sub)
|
||||
except (ValueError, TypeError):
|
||||
raise self.error.credentials_error(detail="Jwt token is incorrect")
|
||||
except (ValueError, TypeError) as e:
|
||||
raise self.error.credentials_error(detail="Jwt token is incorrect") from e
|
||||
|
||||
|
||||
'''old refresh token check'''
|
||||
@@ -122,6 +128,7 @@ class CurrentUser:
|
||||
'''create new refresh token if all the checks are successful'''
|
||||
new_refresh_token, new_jti=self.jwt_service.create_refresh_token({"sub":str(sub)})
|
||||
new_access_token=self.create_access_token(user_id=sub)
|
||||
|
||||
try:
|
||||
new_jti=UUID(new_jti)
|
||||
except (ValueError, TypeError):
|
||||
@@ -152,11 +159,12 @@ class CurrentUser:
|
||||
payload=self.jwt_service.jwt_decode(refresh_token)
|
||||
|
||||
if (jti:=payload.get("jti")) is None:
|
||||
raise self.error.credentials_error(detail="Invalid Refresh Token")
|
||||
raise self.error.credentials_error(detail="Jwt token is incorrect")
|
||||
|
||||
try:
|
||||
jti=UUID(jti)
|
||||
except (ValueError, TypeError):
|
||||
raise self.error.credentials_error(detail="Jwt token is incorrect")
|
||||
except (ValueError, TypeError) as e:
|
||||
raise self.error.credentials_error(detail="Jwt token is incorrect") from e
|
||||
|
||||
current_token = self.jwt_db_actions.get_token_by_id(jti)
|
||||
if current_token is None:
|
||||
|
||||
Reference in New Issue
Block a user