Refactor to manager-centric model, add team calendar for all users

Key changes: - Removed office-centric model (deleted offices.py, office-rules) - Renamed to team-rules, managers are part of their own team - Team calendar visible to all (read-only for employees) - Admins can have a manager assigned
2025-12-02 13:30:04 +00:00
parent 2ad8ba3424
commit 7168fa4b72
30 changed files with 1016 additions and 910 deletions
--- a/app/routes/users.py
+++ b/app/routes/users.py
@@ -11,9 +11,10 @@ import uuid
 import re

 from database.connection import get_db
-from database.models import User, Office, OfficeMembership
+from database.models import User
 from utils.auth_middleware import get_current_user, require_admin
 from services.auth import hash_password, verify_password
+from app import config

 router = APIRouter(prefix="/api/users", tags=["users"])

@@ -24,21 +25,19 @@ class UserCreate(BaseModel):
    password: str
    name: str | None = None
    role: str = "employee"
-    office_id: str | None = None
+    manager_id: str | None = None


 class UserUpdate(BaseModel):
-    email: EmailStr | None = None
    name: str | None = None
    role: str | None = None
-    office_id: str | None = None
+    manager_id: str | None = None
    manager_parking_quota: int | None = None
    manager_spot_prefix: str | None = None


 class ProfileUpdate(BaseModel):
    name: str | None = None
-    office_id: str | None = None


 class SettingsUpdate(BaseModel):
@@ -61,44 +60,86 @@ class UserResponse(BaseModel):
    email: str
    name: str | None
    role: str
-    office_id: str | None
+    manager_id: str | None = None
+    manager_name: str | None = None
    manager_parking_quota: int | None = None
    manager_spot_prefix: str | None = None
+    managed_user_count: int | None = None
+    is_ldap_user: bool = False
+    is_ldap_admin: bool = False
    created_at: str | None

    class Config:
        from_attributes = True


+def user_to_response(user: User, db: Session) -> dict:
+    """Convert user to response dict with computed fields"""
+    # Get manager name if user has a manager
+    manager_name = None
+    if user.manager_id:
+        manager = db.query(User).filter(User.id == user.manager_id).first()
+        if manager:
+            manager_name = manager.name
+
+    # Count managed users if this user is a manager
+    managed_user_count = None
+    if user.role == "manager":
+        managed_user_count = db.query(User).filter(User.manager_id == user.id).count()
+
+    # Determine if user is LDAP-managed
+    is_ldap_user = config.AUTHELIA_ENABLED and user.password_hash is None
+    is_ldap_admin = is_ldap_user and user.role == "admin"
+
+    return {
+        "id": user.id,
+        "email": user.email,
+        "name": user.name,
+        "role": user.role,
+        "manager_id": user.manager_id,
+        "manager_name": manager_name,
+        "manager_parking_quota": user.manager_parking_quota,
+        "manager_spot_prefix": user.manager_spot_prefix,
+        "managed_user_count": managed_user_count,
+        "is_ldap_user": is_ldap_user,
+        "is_ldap_admin": is_ldap_admin,
+        "created_at": user.created_at
+    }
+
+
 # Admin Routes
-@router.get("", response_model=List[UserResponse])
+@router.get("")
 def list_users(db: Session = Depends(get_db), user=Depends(require_admin)):
    """List all users (admin only)"""
    users = db.query(User).all()
-    return users
+    return [user_to_response(u, db) for u in users]


-@router.get("/{user_id}", response_model=UserResponse)
+@router.get("/{user_id}")
 def get_user(user_id: str, db: Session = Depends(get_db), user=Depends(require_admin)):
    """Get user by ID (admin only)"""
    target = db.query(User).filter(User.id == user_id).first()
    if not target:
        raise HTTPException(status_code=404, detail="User not found")
-    return target
+    return user_to_response(target, db)


-@router.post("", response_model=UserResponse)
+@router.post("")
 def create_user(data: UserCreate, db: Session = Depends(get_db), user=Depends(require_admin)):
-    """Create new user (admin only)"""
+    """Create new user (admin only) - only for non-LDAP mode"""
+    if config.AUTHELIA_ENABLED:
+        raise HTTPException(status_code=400, detail="User creation disabled in LDAP mode. Users are created on first login.")
+
    if db.query(User).filter(User.email == data.email).first():
        raise HTTPException(status_code=400, detail="Email already registered")

    if data.role not in ["admin", "manager", "employee"]:
        raise HTTPException(status_code=400, detail="Invalid role")

-    if data.office_id:
-        if not db.query(Office).filter(Office.id == data.office_id).first():
-            raise HTTPException(status_code=404, detail="Office not found")
+    if data.manager_id:
+        manager = db.query(User).filter(User.id == data.manager_id).first()
+        if not manager or manager.role != "manager":
+            raise HTTPException(status_code=400, detail="Invalid manager")

    new_user = User(
        id=str(uuid.uuid4()),
@@ -106,42 +147,61 @@ def create_user(data: UserCreate, db: Session = Depends(get_db), user=Depends(re
        password_hash=hash_password(data.password),
        name=data.name,
        role=data.role,
-        office_id=data.office_id,
+        manager_id=data.manager_id,
        created_at=datetime.utcnow().isoformat()
    )

    db.add(new_user)
    db.commit()
    db.refresh(new_user)
-    return new_user
+    return user_to_response(new_user, db)


-@router.put("/{user_id}", response_model=UserResponse)
+@router.put("/{user_id}")
 def update_user(user_id: str, data: UserUpdate, db: Session = Depends(get_db), user=Depends(require_admin)):
    """Update user (admin only)"""
    target = db.query(User).filter(User.id == user_id).first()
    if not target:
        raise HTTPException(status_code=404, detail="User not found")

-    if data.email is not None:
-        existing = db.query(User).filter(User.email == data.email, User.id != user_id).first()
-        if existing:
-            raise HTTPException(status_code=400, detail="Email already in use")
-        target.email = data.email
+    # Check if user is LDAP-managed
+    is_ldap_user = config.AUTHELIA_ENABLED and target.password_hash is None
+    is_ldap_admin = is_ldap_user and target.role == "admin"

+    # Name update - blocked for LDAP users
    if data.name is not None:
+        if is_ldap_user:
+            raise HTTPException(status_code=400, detail="Name is managed by LDAP")
        target.name = data.name

+    # Role update
    if data.role is not None:
        if data.role not in ["admin", "manager", "employee"]:
            raise HTTPException(status_code=400, detail="Invalid role")
+        # Can't change admin role for LDAP admins (they get admin from parking_admins group)
+        if is_ldap_admin and data.role != "admin":
+            raise HTTPException(status_code=400, detail="Admin role is managed by LDAP group (parking_admins)")
+        # If changing from manager to another role, check for managed users
+        if target.role == "manager" and data.role != "manager":
+            managed_count = db.query(User).filter(User.manager_id == user_id).count()
+            if managed_count > 0:
+                raise HTTPException(status_code=400, detail=f"Cannot change role: {managed_count} users are assigned to this manager")
+            # Clear manager-specific fields
+            target.manager_parking_quota = 0
+            target.manager_spot_prefix = None
        target.role = data.role

-    if data.office_id is not None:
-        if data.office_id and not db.query(Office).filter(Office.id == data.office_id).first():
-            raise HTTPException(status_code=404, detail="Office not found")
-        target.office_id = data.office_id
+    # Manager assignment (any user including admins can be assigned to a manager)
+    if data.manager_id is not None:
+        if data.manager_id:
+            manager = db.query(User).filter(User.id == data.manager_id).first()
+            if not manager or manager.role != "manager":
+                raise HTTPException(status_code=400, detail="Invalid manager")
+            if data.manager_id == user_id:
+                raise HTTPException(status_code=400, detail="User cannot be their own manager")
+        target.manager_id = data.manager_id if data.manager_id else None

+    # Manager-specific fields
    if data.manager_parking_quota is not None:
        if target.role != "manager":
            raise HTTPException(status_code=400, detail="Parking quota only for managers")
@@ -166,7 +226,7 @@ def update_user(user_id: str, data: UserUpdate, db: Session = Depends(get_db), u
    target.updated_at = datetime.utcnow().isoformat()
    db.commit()
    db.refresh(target)
-    return target
+    return user_to_response(target, db)


@router.delete("/{user_id}")
@@ -179,60 +239,53 @@ def delete_user(user_id: str, db: Session = Depends(get_db), current_user=Depend
    if not target:
        raise HTTPException(status_code=404, detail="User not found")

+    # Check if user is a manager with managed users
+    if target.role == "manager":
+        managed_count = db.query(User).filter(User.manager_id == user_id).count()
+        if managed_count > 0:
+            raise HTTPException(status_code=400, detail=f"Cannot delete: {managed_count} users are assigned to this manager")
+
    db.delete(target)
    db.commit()
    return {"message": "User deleted"}


 # Self-service Routes
-@router.get("/me/managed-offices")
-def get_managed_offices(db: Session = Depends(get_db), current_user=Depends(get_current_user)):
-    """Get offices the current user manages"""
-    if current_user.role == "admin":
-        offices = db.query(Office).all()
-        return {"role": "admin", "offices": [{"id": o.id, "name": o.name} for o in offices]}
-
-    if current_user.role == "manager":
-        memberships = db.query(OfficeMembership).filter(
-            OfficeMembership.user_id == current_user.id
-        ).all()
-        office_ids = [m.office_id for m in memberships]
-        offices = db.query(Office).filter(Office.id.in_(office_ids)).all()
-        return {"role": "manager", "offices": [{"id": o.id, "name": o.name} for o in offices]}
-
-    if current_user.office_id:
-        office = db.query(Office).filter(Office.id == current_user.office_id).first()
-        if office:
-            return {"role": "employee", "offices": [{"id": office.id, "name": office.name}]}
-
-    return {"role": current_user.role, "offices": []}
-
-
@router.get("/me/profile")
-def get_profile(current_user=Depends(get_current_user)):
+def get_profile(db: Session = Depends(get_db), current_user=Depends(get_current_user)):
    """Get current user's profile"""
+    is_ldap_user = config.AUTHELIA_ENABLED and current_user.password_hash is None
+
+    # Get manager name
+    manager_name = None
+    if current_user.manager_id:
+        manager = db.query(User).filter(User.id == current_user.manager_id).first()
+        if manager:
+            manager_name = manager.name
+
    return {
        "id": current_user.id,
        "email": current_user.email,
        "name": current_user.name,
        "role": current_user.role,
-        "office_id": current_user.office_id
+        "manager_id": current_user.manager_id,
+        "manager_name": manager_name,
+        "is_ldap_user": is_ldap_user
    }


@router.put("/me/profile")
 def update_profile(data: ProfileUpdate, db: Session = Depends(get_db), current_user=Depends(get_current_user)):
-    """Update current user's profile"""
+    """Update current user's profile (limited fields)"""
+    is_ldap_user = config.AUTHELIA_ENABLED and current_user.password_hash is None
+
    if data.name is not None:
+        if is_ldap_user:
+            raise HTTPException(status_code=400, detail="Name is managed by LDAP")
        current_user.name = data.name
+        current_user.updated_at = datetime.utcnow().isoformat()
+        db.commit()

-    if data.office_id is not None:
-        if data.office_id and not db.query(Office).filter(Office.id == data.office_id).first():
-            raise HTTPException(status_code=404, detail="Office not found")
-        current_user.office_id = data.office_id if data.office_id else None
-
-    current_user.updated_at = datetime.utcnow().isoformat()
-    db.commit()
    return {"message": "Profile updated"}


@@ -292,7 +345,10 @@ def update_settings(data: SettingsUpdate, db: Session = Depends(get_db), current

@router.post("/me/change-password")
 def change_password(data: ChangePasswordRequest, db: Session = Depends(get_db), current_user=Depends(get_current_user)):
-    """Change current user's password"""
+    """Change current user's password (not available in LDAP mode)"""
+    if config.AUTHELIA_ENABLED and current_user.password_hash is None:
+        raise HTTPException(status_code=400, detail="Password is managed by LDAP")
+
    if not verify_password(data.current_password, current_user.password_hash):
        raise HTTPException(status_code=400, detail="Current password is incorrect")