Apply responsability segregation in user route and jwt_manager

Author: ArmCM

routers/user.py

@@ -1,39 +1,40 @@
-from fastapi import status,APIRouter
+from fastapi import status, APIRouter
 from fastapi.responses import JSONResponse
 from utils.jwt_manager import create_token
-from schemas.user import User,UserBase,UserCreate
+from schemas.user import User, UserBase, UserCreate
 from config.database import Session
 from services.user import UserService
 from services.auth import Auth
 
 user_router = APIRouter()
+db = Session()
+
+
+@user_router.post('/users', tags=['Auth'], response_model=User, status_code=status.HTTP_200_OK)
+def create_user(user: UserCreate):
+    check_user_exists(user)
 
-@user_router.post('/users',tags=['Auth'],response_model=User,status_code=status.HTTP_200_OK)
-def create_user(user:UserCreate):
-    
-    db = Session()
-    
-    result = UserService(db).get_user_by_email(email=user.email)
-    
-    if result:
-    
-        return JSONResponse(status_code=status.HTTP_400_BAD_REQUEST,content={"message":"User already exists"})
-    
     UserService(db).create_user(user)
-    
-    return JSONResponse(status_code=status.HTTP_200_OK,content={"message":"User created"})
-    
-
-@user_router.post('/login',tags=['Auth'],status_code=status.HTTP_200_OK)
-def login(user:UserCreate):
-    
-    db = Session()
-    result = UserService(db).get_user_by_email(email=user.email)
-    
-    if not (result and Auth().verify_password(user.password,result.password)):
-        
-        return JSONResponse(status_code=status.HTTP_401_UNAUTHORIZED,content={"message":"Unauthorized"})
-    
-    token:str = create_token(user.dict())
-    
-    return JSONResponse(status_code=status.HTTP_200_OK,content=token)
+
+    return JSONResponse(status_code=status.HTTP_200_OK, content={"message": "User created"})
+
+
+def check_user_exists(user):
+    if UserService(db).get_user_by_email(email=user.email):
+        return JSONResponse(status_code=status.HTTP_400_BAD_REQUEST, content={"message": "User already exists"})
+
+
+@user_router.post('/login', tags=['Auth'], status_code=status.HTTP_200_OK)
+def login(user: UserCreate):
+    validate_password(user)
+
+    token: str = create_token(user.dict())
+
+    return JSONResponse(status_code=status.HTTP_200_OK, content=token)
+
+
+def validate_password(user):
+    user_found = UserService(db).get_user_by_email(email=user.email)
+
+    if not (user_found and Auth().verify_password(user.password, user_found.password)):
+        return JSONResponse(status_code=status.HTTP_401_UNAUTHORIZED, content={"message": "Unauthorized"})

utils/jwt_manager.py

@@ -1,21 +1,28 @@
-from jwt import encode,decode
-from datetime import datetime,timedelta
+from jwt import encode, decode
+from datetime import datetime, timedelta
 from utils.settings import Settings
 
 settings = Settings()
 
-def create_token(data:dict) -> dict:
-    payload = expire_token(data)
-    token:str = encode(payload,key=settings.MY_SECRET_KEY,algorithm="HS256")
+
+def create_token(data: dict) -> dict:
+    payload = encode_payload(data)
+
+    token: str = encode(payload, key=settings.MY_SECRET_KEY, algorithm="HS256")
     return token
 
-def validate_token(token:str) -> dict:
-    data:dict = decode(token,key=settings.MY_SECRET_KEY,algorithms=["HS256"]) 
+
+def validate_token(token: str) -> dict:
+    data: dict = decode(token, key=settings.MY_SECRET_KEY, algorithms=["HS256"])
     return data
 
-def expire_token(data:dict):
+
+def calculate_token_expiration():
+    return datetime.utcnow() + timedelta(minutes=settings.TOKEN_EXPIRE_MINUTES)
+
+
+def encode_payload(data: dict):
     to_encode = data.copy()
-    token_expires = timedelta(minutes=settings.TOKEN_EXPIRE_MINUTES)
-    expire = datetime.utcnow() + token_expires
-    to_encode.update({'exp':expire})
-    return to_encode
+    to_encode['exp'] = calculate_token_expiration()
+
+    return to_encode