Last push before leaving

src/crud/__init__.py

@@ -7,9 +7,6 @@ CRUD function directly from `src.crud` instead of the specific submodule,
 keeping the router imports clean.
 """
 
-# Import utility functions first
-from .utils import hash_password, verify_password
-
 from .students import (
     create_student,
     get_all_students,
@@ -25,6 +22,7 @@ from .users import (
     update_user_tag_id,
     get_user_by_id,
     delete_user,
+    hash_password,  # Import hash_password from users module
 )
 from .devices import (
     get_device_by_id_str,
@@ -44,11 +42,16 @@ from .tag_linking import (
     delete_pending_link,
 )
 
-# Export all functions (optional, for explicit imports)
+# Export all functions
 __all__ = [
-    # Utilities
+    # Users
+    'create_user',
+    'get_user_by_username',
+    'get_user_by_tag_id',
+    'update_user_tag_id',
+    'get_user_by_id',
+    'delete_user',
     'hash_password',
-    'verify_password',
     # Students
     'create_student',
     'get_all_students',
@@ -56,13 +59,6 @@ __all__ = [
     'get_student_by_tag_id',
     'update_student_tag_id',
     'delete_student',
-    # Users
-    'create_user',
-    'get_user_by_username',
-    'get_user_by_tag_id',
-    'update_user_tag_id',
-    'get_user_by_id',
-    'delete_user',
     # Devices
     'get_device_by_id_str',
     'get_device_by_api_key',

src/crud/users.py

@@ -3,9 +3,13 @@ CRUD operations for Users (Admins, Staff).
 """
 from sqlalchemy.orm import Session
 from fastapi import HTTPException, status
+import bcrypt
 
 from src import models
-from src.crud import hash_password  # Import from crud module instead of auth
+
+def hash_password(password: str) -> str:
+    """Hashes a password using bcrypt."""
+    return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
 
 def get_user_by_username(db: Session, username: str) -> models.User | None:
     """
@@ -25,12 +29,39 @@ def get_all_users(db: Session, skip: int = 0, limit: int = 100) -> list[models.U
     """
     return db.query(models.User).offset(skip).limit(limit).all()
 
-def get_user_by_tag_id(db:Session, tag_id:int) -> models.User | None:
+def get_user_by_tag_id(db: Session, tag_id: str) -> models.User | None:
     """
     Retrieves a user by their TAG_ID.
     """
     return db.query(models.User).filter(models.User.tag_id == tag_id).first()
 
+def update_user_tag_id(db: Session, username: str, new_tag_id: str) -> models.User:
+    """
+    Updates a user's tag_id.
+    """
+    # Check if the new tag_id is already in use
+    existing_user_with_tag = get_user_by_tag_id(db, new_tag_id)
+    if existing_user_with_tag:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Tag ID '{new_tag_id}' is already assigned to another user."
+        )
+    
+    # Get the user to update
+    user = get_user_by_username(db, username)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"User '{username}' not found."
+        )
+    
+    # Update the tag_id
+    user.tag_id = new_tag_id
+    db.commit()
+    db.refresh(user)
+    
+    return user
+
 def create_user(db: Session, user_data: models.UserCreate) -> models.User:
     """
     Creates a new user account.
@@ -43,8 +74,17 @@ def create_user(db: Session, user_data: models.UserCreate) -> models.User:
             detail=f"Username '{user_data.username}' is already registered."
         )
 
+    # Check if tag_id is already in use (if provided)
+    if user_data.tag_id:
+        existing_tag_user = get_user_by_tag_id(db, user_data.tag_id)
+        if existing_tag_user:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Tag ID '{user_data.tag_id}' is already assigned to another user."
+            )
+
     # Hash the password
-    hashed_password = hash_password(user_data.password)  # Use the imported function
+    hashed_password = hash_password(user_data.password)
     
     # Create new user
     db_user = models.User(
@@ -74,6 +114,24 @@ def update_user(db: Session, user_id: int, user_update: models.UserCreate) -> mo
             detail=f"User with ID {user_id} not found."
         )
 
+    # Check if new username is already taken (if changed)
+    if user_update.username != db_user.username:
+        existing_user = get_user_by_username(db, user_update.username)
+        if existing_user:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Username '{user_update.username}' is already taken."
+            )
+
+    # Check if new tag_id is already in use (if changed and provided)
+    if user_update.tag_id and user_update.tag_id != db_user.tag_id:
+        existing_tag_user = get_user_by_tag_id(db, user_update.tag_id)
+        if existing_tag_user:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Tag ID '{user_update.tag_id}' is already assigned to another user."
+            )
+
     # Update fields
     db_user.username = user_update.username
     db_user.name = user_update.name
@@ -117,7 +175,13 @@ def delete_user(db: Session, username_to_delete: str, current_admin: models.User
 
     # Handle dependencies: set foreign keys to NULL where a user is referenced
     db.query(models.ClearanceStatus).filter(models.ClearanceStatus.cleared_by == user_to_delete.id).update({"cleared_by": None})
-    db.query(models.PendingTagLink).filter(models.PendingTagLink.initiated_by_user_id == user_to_delete.id).delete()
+    
+    # Check if PendingTagLink model exists before trying to delete
+    try:
+        db.query(models.PendingTagLink).filter(models.PendingTagLink.initiated_by_user_id == user_to_delete.id).delete()
+    except AttributeError:
+        # PendingTagLink model doesn't exist, skip this cleanup
+        pass
 
     db.delete(user_to_delete)
     db.commit()