Refactor code

.flake8

@@ -0,0 +1,4 @@
+[flake8]
+max-line-length = 120
+exclude = .git,__pycache__,venv,tests/*
+ignore = F541

.github/workflows/ci.yml

@@ -24,8 +24,17 @@ jobs:
           python -m pip install --upgrade pip  # Upgrade pip to the latest version
           pip install setuptools wheel
           pip install -r requirements.txt  # Install dependencies from requirements.txt
+          pip install -r requirements-dev.txt
           pip install pylint pytest
+
+      - name: Run Flake8
+        run: flake8 .
 
+      - name: Run Black
+        run: black . --check
+
+      - name: Run MyPy
+        run: mypy .
       - name: Run Tests
         run: |
           python -m pytest tests/

.gitignore

@@ -31,4 +31,8 @@ build/
 htmlcov/
 .tox/
 .coverage
-.coverage.*
+.coverage.*
+
+# Reports
+*_report.txt
+reports/

.pre-commit-config.yaml

@@ -0,0 +1,17 @@
+repos:
+  - repo: https://github.com/psf/black
+    rev: 25.1.0 
+    hooks:
+      - id: black
+        language_version: python3.12 
+
+  - repo: https://github.com/PyCQA/flake8
+    rev: '7.2.0'  
+    hooks:
+      - id: flake8
+
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: 'v1.15.0' 
+    hooks:
+      - id: mypy
+

app/auth/router.py

@@ -1,28 +1,34 @@
 from datetime import datetime, timedelta
 from typing import Optional
+
 from fastapi import APIRouter, Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
-from jose import JWTError, jwt
+from pydantic import BaseModel, Field, validator
+from jose import jwt, JWTError
 from sqlalchemy.orm import Session
-from app.database import get_db
-from app.models import User, UserRole
 from passlib.context import CryptContext
-from pydantic import BaseModel, Field, validator
+
+from app.database import get_db
+from app.models import User
+from app.enums import UserRole
 
 router = APIRouter(prefix="/auth", tags=["authentication"])
 
+
+# Schemas
 class UserCreate(BaseModel):
     username: str = Field(..., min_length=3, max_length=50)
     email: str
     password: str
     role: UserRole
 
-    @validator('role')
+    @validator("role")
     def validate_role(cls, v):
         if v not in [UserRole.admin, UserRole.case_worker]:
-            raise ValueError('Role must be either admin or case_worker')
+            raise ValueError("Role must be either admin or case_worker")
         return v
 
+
 class UserResponse(BaseModel):
     username: str
     email: str
@@ -31,121 +37,146 @@ class UserResponse(BaseModel):
     class Config:
         from_attributes = True
 
-# Configuration
-SECRET_KEY = "your-secret-key-here"
-ALGORITHM = "HS256"
-ACCESS_TOKEN_EXPIRE_MINUTES = 30
 
-pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")
+# Security Service
+class SecurityService:
+    pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
-def verify_password(plain_password: str, hashed_password: str) -> bool:
-    return pwd_context.verify(plain_password, hashed_password)
+    def verify_password(self, plain_password: str, hashed_password: str) -> bool:
+        return self.pwd_context.verify(plain_password, hashed_password)
 
-def get_password_hash(password: str) -> str:
-    return pwd_context.hash(password)
+    def get_password_hash(self, password: str) -> str:
+        return self.pwd_context.hash(password)
 
-def authenticate_user(db: Session, username: str, password: str) -> Optional[User]:
-    user = db.query(User).filter(User.username == username).first()
-    if not user or not verify_password(password, user.hashed_password):
-        return None
-    return user
 
-def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
-    to_encode = data.copy()
-    if expires_delta:
-        expire = datetime.utcnow() + expires_delta
-    else:
-        expire = datetime.utcnow() + timedelta(minutes=15)
-    to_encode.update({"exp": expire})
-    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
-    return encoded_jwt
+security = SecurityService()
+
+
+# Token Service
+class TokenService:
+    SECRET_KEY = "your-secret-key-here"
+    ALGORITHM = "HS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+    def create_access_token(
+        self, data: dict, expires_delta: Optional[timedelta] = None
+    ):
+        to_encode = data.copy()
+        expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
+        to_encode.update({"exp": expire})
+        return jwt.encode(to_encode, self.SECRET_KEY, algorithm=self.ALGORITHM)
+
+    def decode_token(self, token: str):
+        try:
+            return jwt.decode(token, self.SECRET_KEY, algorithms=[self.ALGORITHM])
+        except JWTError:
+            return None
+
+
+token_service = TokenService()
+
+
+# User Service
+class UserService:
+    def authenticate_user(
+        self, db: Session, username: str, password: str
+    ) -> Optional[User]:
+        user = db.query(User).filter(User.username == username).first()
+        if not user or not security.verify_password(password, user.hashed_password):
+            return None
+        return user
+
+    def create_user(self, db: Session, user_data: UserCreate) -> User:
+        if db.query(User).filter(User.username == user_data.username).first():
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Username already registered",
+            )
+
+        if db.query(User).filter(User.email == user_data.email).first():
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Email already registered",
+            )
+
+        db_user = User(
+            username=user_data.username,
+            email=user_data.email,
+            hashed_password=security.get_password_hash(user_data.password),
+            role=user_data.role,
+        )
+
+        try:
+            db.add(db_user)
+            db.commit()
+            db.refresh(db_user)
+            return db_user
+        except Exception as e:
+            db.rollback()
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail=str(e),
+            )
+
+
+user_service = UserService()
+
+
+# Auth Dependencies
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")
+
 
 async def get_current_user(
-    token: str = Depends(oauth2_scheme),
-    db: Session = Depends(get_db)
+    token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)
 ) -> User:
     credentials_exception = HTTPException(
         status_code=status.HTTP_401_UNAUTHORIZED,
         detail="Could not validate credentials",
         headers={"WWW-Authenticate": "Bearer"},
     )
-    try:
-        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
-        username: str = payload.get("sub")
-        if username is None:
-            raise credentials_exception
-    except JWTError:
+    payload = token_service.decode_token(token)
+    username = payload.get("sub") if payload else None
+    if not username:
         raise credentials_exception
-    
+
     user = db.query(User).filter(User.username == username).first()
     if user is None:
         raise credentials_exception
     return user
 
+
 def get_admin_user(current_user: User = Depends(get_current_user)):
     if current_user.role != UserRole.admin:
         raise HTTPException(
             status_code=status.HTTP_403_FORBIDDEN,
-            detail="Only admin users can perform this operation"
+            detail="Only admin users can perform this operation",
         )
     return current_user
 
+
+# Routes
 @router.post("/token")
 async def login_for_access_token(
-    form_data: OAuth2PasswordRequestForm = Depends(),
-    db: Session = Depends(get_db)
+    form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)
 ):
-    user = authenticate_user(db, form_data.username, form_data.password)
+    user = user_service.authenticate_user(db, form_data.username, form_data.password)
     if not user:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="Incorrect username or password",
             headers={"WWW-Authenticate": "Bearer"},
         )
-    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
-    access_token = create_access_token(
+    access_token_expires = timedelta(minutes=token_service.ACCESS_TOKEN_EXPIRE_MINUTES)
+    access_token = token_service.create_access_token(
         data={"sub": user.username}, expires_delta=access_token_expires
     )
     return {"access_token": access_token, "token_type": "bearer"}
 
+
 @router.post("/users", response_model=UserResponse)
 async def create_user(
     user_data: UserCreate,
     current_user: User = Depends(get_admin_user),
-    db: Session = Depends(get_db)
+    db: Session = Depends(get_db),
 ):
-    """Create a new user (admin only)"""
-    # Check if username exists
-    if db.query(User).filter(User.username == user_data.username).first():
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail="Username already registered"
-        )
-
-    # Check if email exists
-    if db.query(User).filter(User.email == user_data.email).first():
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail="Email already registered"
-        )
-
-    # Create new user
-    db_user = User(
-        username=user_data.username,
-        email=user_data.email,
-        hashed_password=get_password_hash(user_data.password),
-        role=user_data.role
-    )
-
-    try:
-        db.add(db_user)
-        db.commit()
-        db.refresh(db_user)
-        return db_user
-    except Exception as e:
-        db.rollback()
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=str(e)
-        )
+    return user_service.create_user(db, user_data)