From e953e09fee8f66fc770ed91c44190747fb47fec9 Mon Sep 17 00:00:00 2001 From: Borgia Leiva Date: Tue, 3 Feb 2026 10:24:00 +0100 Subject: [PATCH] refactored reused code blocks to use helper function in routers.auth --- routers/auth.py | 6 ++- routers/projects.py | 124 +++----------------------------------------- routers/users.py | 6 +-- 3 files changed, 14 insertions(+), 122 deletions(-) diff --git a/routers/auth.py b/routers/auth.py index 6f294fa..b6c8586 100644 --- a/routers/auth.py +++ b/routers/auth.py @@ -16,7 +16,9 @@ SECRET_KEY = os.getenv("SECRET_KEY", "your-secret-key-change-this-in-production" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 1440 # 24 hours -def create_access_token(data: dict, expires_delta: timedelta | None = None): +def create_access_token(data: dict, expires_delta: timedelta | None = None): + """Create a JWT token""" + to_encode = data.copy() if expires_delta: expire = datetime.now(timezone.utc) + expires_delta @@ -27,7 +29,7 @@ def create_access_token(data: dict, expires_delta: timedelta | None = None): encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt -def check_for_valid_token(request: Request, db: db_dependency) -> models.User : +def get_user_from_jwt(request: Request, db: db_dependency) -> models.User : """Helper function to check for valid JWT token in cookies""" token = request.cookies.get("access_token") if not token: diff --git a/routers/projects.py b/routers/projects.py index 065b188..226b26e 100644 --- a/routers/projects.py +++ b/routers/projects.py @@ -12,6 +12,7 @@ import schemas.projects as projects_schemas import schemas.users as users_schemas from models import Project +from routers.auth import get_user_from_jwt router = APIRouter(prefix="/projects", tags=["projects"]) @@ -23,29 +24,8 @@ router = APIRouter(prefix="/projects", tags=["projects"]) def get_projects(db: db_dependency, request: Request): """Get a user's projects""" - ## User retrieval from JWT token in cookies - get_token = request.cookies.get("access_token") - if not get_token: - raise HTTPException( - status_code=401, - detail="Not logged in" - ) - - try: - payload = jwt.decode(get_token, auth.SECRET_KEY, algorithms=[auth.ALGORITHM]) - user_id: str = str(payload.get("sub")) - if user_id is None: - request.cookies.clear() ## User in cookies not found, clear cookies - raise HTTPException( - status_code=401, - detail="Not logged in" - ) - except JWTError: - request.cookies.clear() ## Probably an invalid token, clear cookies - raise HTTPException( - status_code=401, - detail="Could not validate credentials" - ) + user = get_user_from_jwt(request, db) + user_id = getattr(user, "id") ## fetching projects for the user projects = db.query(Project).join(Project.users).filter(getattr(users_schemas.UserBase, "id") == int(user_id)).all() @@ -56,29 +36,7 @@ def get_projects(db: db_dependency, request: Request): def get_project(project_id: int, request:Request, db: db_dependency): """Get a project by ID""" - get_token = request.cookies.get("access_token") - if not get_token: - raise HTTPException( - status_code=401, - detail="Not logged in" - ) - - try: - payload = jwt.decode(get_token, auth.SECRET_KEY, algorithms=[auth.ALGORITHM]) - user_id: str = str(payload.get("sub")) - if user_id is None: - request.cookies.clear() ## User in cookies not found, clear cookies - raise HTTPException( - status_code=401, - detail="Not logged in" - ) - except JWTError: - request.cookies.clear() ## Probably an invalid token, clear cookies - raise HTTPException( - status_code=401, - detail="Could not validate credentials" - ) - user = db.query(users_schemas.UserBase).filter(getattr(users_schemas.UserBase, "id") == user_id).first() + user = get_user_from_jwt(request, db) db_project = db.query(projects_schemas.ProjectBase).filter(getattr(projects_schemas.ProjectBase, "id") == project_id).first() if db_project is None: @@ -92,30 +50,7 @@ def get_project(project_id: int, request:Request, db: db_dependency): def get_project_users(project_id: int, request:Request, db: db_dependency): """Get users from a specified project""" - get_token = request.cookies.get("access_token") - if not get_token: - raise HTTPException( - status_code=401, - detail="Not logged in" - ) - - try: - payload = jwt.decode(get_token, auth.SECRET_KEY, algorithms=[auth.ALGORITHM]) - user_id: str = str(payload.get("sub")) - if user_id is None: - request.cookies.clear() ## User in cookies not found, clear cookies - raise HTTPException( - status_code=401, - detail="Not logged in" - ) - except JWTError: - request.cookies.clear() ## Probably an invalid token, clear cookies - raise HTTPException( - status_code=401, - detail="Could not validate credentials" - ) - user = db.query(users_schemas.UserBase).filter(getattr(users_schemas.UserBase, "id") == user_id).first() - + user = get_user_from_jwt(request, db) db_project = db.query(projects_schemas.ProjectBase).filter(getattr(projects_schemas.ProjectBase, "id") == project_id).first() if db_project is None: raise HTTPException(status_code=404, detail="Project not found") @@ -128,29 +63,7 @@ def get_project_users(project_id: int, request:Request, db: db_dependency): def get_project_tasks(project_id: int, request:Request, db: db_dependency): """Get tasks from a specified project""" - get_token = request.cookies.get("access_token") - if not get_token: - raise HTTPException( - status_code=401, - detail="Not logged in" - ) - - try: - payload = jwt.decode(get_token, auth.SECRET_KEY, algorithms=[auth.ALGORITHM]) - user_id: str = str(payload.get("sub")) - if user_id is None: - request.cookies.clear() ## User in cookies not found, clear cookies - raise HTTPException( - status_code=401, - detail="Not logged in" - ) - except JWTError: - request.cookies.clear() ## Probably an invalid token, clear cookies - raise HTTPException( - status_code=401, - detail="Could not validate credentials" - ) - user = db.query(users_schemas.UserBase).filter(getattr(users_schemas.UserBase, "id") == user_id).first() + user = get_user_from_jwt(request, db) db_project = db.query(projects_schemas.ProjectBase).filter(getattr(projects_schemas.ProjectBase, "id") == project_id).first() if db_project is None: @@ -164,30 +77,7 @@ def get_project_tasks(project_id: int, request:Request, db: db_dependency): def create_project(project: projects_schemas.ProjectCreate, request:Request, db: db_dependency): """Create a new project""" - get_token = request.cookies.get("access_token") - if not get_token: - raise HTTPException( - status_code=401, - detail="Not logged in" - ) - - try: - payload = jwt.decode(get_token, auth.SECRET_KEY, algorithms=[auth.ALGORITHM]) - user_id: str = str(payload.get("sub")) - if user_id is None: - request.cookies.clear() ## User in cookies not found, clear cookies - raise HTTPException( - status_code=401, - detail="Not logged in" - ) - except JWTError: - request.cookies.clear() ## Probably an invalid token, clear cookies - raise HTTPException( - status_code=401, - detail="Could not validate credentials" - ) - user = db.query(users_schemas.UserBase).filter(getattr(users_schemas.UserBase, "id") == user_id).first() - + user = get_user_from_jwt(request, db) db_project = projects_schemas.ProjectCreate( name=project.name, description=project.description, diff --git a/routers/users.py b/routers/users.py index 74f6128..3e29d33 100644 --- a/routers/users.py +++ b/routers/users.py @@ -9,7 +9,7 @@ import models from routers import auth import schemas.users as users import schemas.projects as projects -from routers.auth import check_for_valid_token +from routers.auth import get_user_from_jwt from pyargon2 import hash @@ -20,7 +20,7 @@ router = APIRouter(prefix="/users", tags=["users"]) def read_user(user_id: int, db: db_dependency, request:Request): """Get a user by ID""" - check_for_valid_token(request, db) + get_user_from_jwt(request, db) db_user = db.query(models.User).filter(models.User.id == user_id).first() if db_user is None: @@ -31,7 +31,7 @@ def read_user(user_id: int, db: db_dependency, request:Request): def read_projects_from_user(user_id: int, db: db_dependency, request: Request): """Get projects assigned to a user""" - check_for_valid_token(request, db) + get_user_from_jwt(request, db) db_user = db.query(models.User).filter(models.User.id == user_id).first() if db_user is None: