feat: implement admin-users management with super admin restriction
- Add /admin/admin-users routes for managing admin users (super admin only) - Remove vendor role from user creation form (vendors created via company hierarchy) - Add admin-users.html and admin-user-detail.html templates - Add admin-users.js and admin-user-detail.js for frontend logic - Move database operations to admin_platform_service (list, get, create, delete, toggle status) - Update sidebar to show Admin Users section only for super admins - Add isSuperAdmin computed property to init-alpine.js - Fix /api/v1 prefix issues in JS files (apiClient already adds prefix) - Update architecture rule JS-012 to catch more variable patterns (url, endpoint, path) - Replace inline SVGs with $icon() helper in select-platform.html Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -4,9 +4,11 @@ Admin user management endpoints (Super Admin only).
|
||||
|
||||
This module provides endpoints for:
|
||||
- Listing all admin users with their platform assignments
|
||||
- Creating platform admins
|
||||
- Creating platform admins and super admins
|
||||
- Assigning/removing platform access
|
||||
- Promoting/demoting super admin status
|
||||
- Toggling admin status
|
||||
- Deleting admin users
|
||||
"""
|
||||
|
||||
import logging
|
||||
@@ -18,6 +20,7 @@ from sqlalchemy.orm import Session
|
||||
|
||||
from app.api.deps import get_current_super_admin, get_current_super_admin_api
|
||||
from app.core.database import get_db
|
||||
from app.exceptions import ValidationException
|
||||
from app.services.admin_platform_service import admin_platform_service
|
||||
from models.database.user import User
|
||||
|
||||
@@ -65,15 +68,16 @@ class AdminUserListResponse(BaseModel):
|
||||
total: int
|
||||
|
||||
|
||||
class CreatePlatformAdminRequest(BaseModel):
|
||||
"""Request to create a new platform admin."""
|
||||
class CreateAdminUserRequest(BaseModel):
|
||||
"""Request to create a new admin user (platform admin or super admin)."""
|
||||
|
||||
email: EmailStr
|
||||
username: str
|
||||
password: str
|
||||
first_name: Optional[str] = None
|
||||
last_name: Optional[str] = None
|
||||
platform_ids: list[int]
|
||||
is_super_admin: bool = False
|
||||
platform_ids: list[int] = []
|
||||
|
||||
|
||||
class AssignPlatformRequest(BaseModel):
|
||||
@@ -89,147 +93,12 @@ class ToggleSuperAdminRequest(BaseModel):
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# ENDPOINTS
|
||||
# HELPER FUNCTIONS
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@router.get("", response_model=AdminUserListResponse)
|
||||
def list_admin_users(
|
||||
skip: int = Query(0, ge=0),
|
||||
limit: int = Query(100, ge=1, le=500),
|
||||
include_super_admins: bool = Query(True),
|
||||
db: Session = Depends(get_db),
|
||||
current_admin: User = Depends(get_current_super_admin),
|
||||
):
|
||||
"""
|
||||
List all admin users with their platform assignments.
|
||||
|
||||
Super admin only.
|
||||
"""
|
||||
from sqlalchemy.orm import joinedload
|
||||
|
||||
query = db.query(User).filter(User.role == "admin")
|
||||
|
||||
if not include_super_admins:
|
||||
query = query.filter(User.is_super_admin == False)
|
||||
|
||||
total = query.count()
|
||||
|
||||
admins = (
|
||||
query.options(joinedload(User.admin_platforms))
|
||||
.offset(skip)
|
||||
.limit(limit)
|
||||
.all()
|
||||
)
|
||||
|
||||
admin_responses = []
|
||||
for admin in admins:
|
||||
assignments = []
|
||||
if not admin.is_super_admin:
|
||||
for ap in admin.admin_platforms:
|
||||
if ap.is_active and ap.platform:
|
||||
assignments.append(
|
||||
PlatformAssignmentResponse(
|
||||
platform_id=ap.platform_id,
|
||||
platform_code=ap.platform.code,
|
||||
platform_name=ap.platform.name,
|
||||
is_active=ap.is_active,
|
||||
)
|
||||
)
|
||||
|
||||
admin_responses.append(
|
||||
AdminUserResponse(
|
||||
id=admin.id,
|
||||
email=admin.email,
|
||||
username=admin.username,
|
||||
first_name=admin.first_name,
|
||||
last_name=admin.last_name,
|
||||
is_active=admin.is_active,
|
||||
is_super_admin=admin.is_super_admin,
|
||||
platform_assignments=assignments,
|
||||
)
|
||||
)
|
||||
|
||||
return AdminUserListResponse(admins=admin_responses, total=total)
|
||||
|
||||
|
||||
@router.post("", response_model=AdminUserResponse)
|
||||
def create_platform_admin(
|
||||
request: CreatePlatformAdminRequest,
|
||||
db: Session = Depends(get_db),
|
||||
current_admin: User = Depends(get_current_super_admin_api),
|
||||
):
|
||||
"""
|
||||
Create a new platform admin with platform assignments.
|
||||
|
||||
Super admin only.
|
||||
"""
|
||||
user, assignments = admin_platform_service.create_platform_admin(
|
||||
db=db,
|
||||
email=request.email,
|
||||
username=request.username,
|
||||
password=request.password,
|
||||
platform_ids=request.platform_ids,
|
||||
created_by_user_id=current_admin.id,
|
||||
first_name=request.first_name,
|
||||
last_name=request.last_name,
|
||||
)
|
||||
|
||||
db.commit()
|
||||
|
||||
# Refresh to get relationships
|
||||
db.refresh(user)
|
||||
|
||||
assignment_responses = [
|
||||
PlatformAssignmentResponse(
|
||||
platform_id=ap.platform_id,
|
||||
platform_code=ap.platform.code if ap.platform else "",
|
||||
platform_name=ap.platform.name if ap.platform else "",
|
||||
is_active=ap.is_active,
|
||||
)
|
||||
for ap in user.admin_platforms
|
||||
if ap.is_active
|
||||
]
|
||||
|
||||
logger.info(f"Created platform admin {user.username} by {current_admin.username}")
|
||||
|
||||
return AdminUserResponse(
|
||||
id=user.id,
|
||||
email=user.email,
|
||||
username=user.username,
|
||||
first_name=user.first_name,
|
||||
last_name=user.last_name,
|
||||
is_active=user.is_active,
|
||||
is_super_admin=user.is_super_admin,
|
||||
platform_assignments=assignment_responses,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/{user_id}", response_model=AdminUserResponse)
|
||||
def get_admin_user(
|
||||
user_id: int = Path(...),
|
||||
db: Session = Depends(get_db),
|
||||
current_admin: User = Depends(get_current_super_admin),
|
||||
):
|
||||
"""
|
||||
Get admin user details with platform assignments.
|
||||
|
||||
Super admin only.
|
||||
"""
|
||||
from sqlalchemy.orm import joinedload
|
||||
|
||||
from app.exceptions import ValidationException
|
||||
|
||||
admin = (
|
||||
db.query(User)
|
||||
.options(joinedload(User.admin_platforms))
|
||||
.filter(User.id == user_id, User.role == "admin")
|
||||
.first()
|
||||
)
|
||||
|
||||
if not admin:
|
||||
raise ValidationException("Admin user not found", field="user_id")
|
||||
|
||||
def _build_admin_response(admin: User) -> AdminUserResponse:
|
||||
"""Build AdminUserResponse from User model."""
|
||||
assignments = []
|
||||
if not admin.is_super_admin:
|
||||
for ap in admin.admin_platforms:
|
||||
@@ -255,6 +124,111 @@ def get_admin_user(
|
||||
)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# ENDPOINTS
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@router.get("", response_model=AdminUserListResponse)
|
||||
def list_admin_users(
|
||||
skip: int = Query(0, ge=0),
|
||||
limit: int = Query(100, ge=1, le=500),
|
||||
include_super_admins: bool = Query(True),
|
||||
db: Session = Depends(get_db),
|
||||
current_admin: User = Depends(get_current_super_admin),
|
||||
):
|
||||
"""
|
||||
List all admin users with their platform assignments.
|
||||
|
||||
Super admin only.
|
||||
"""
|
||||
admins, total = admin_platform_service.list_admin_users(
|
||||
db=db,
|
||||
skip=skip,
|
||||
limit=limit,
|
||||
include_super_admins=include_super_admins,
|
||||
)
|
||||
|
||||
admin_responses = [_build_admin_response(admin) for admin in admins]
|
||||
|
||||
return AdminUserListResponse(admins=admin_responses, total=total)
|
||||
|
||||
|
||||
@router.post("", response_model=AdminUserResponse)
|
||||
def create_admin_user(
|
||||
request: CreateAdminUserRequest,
|
||||
db: Session = Depends(get_db),
|
||||
current_admin: User = Depends(get_current_super_admin_api),
|
||||
):
|
||||
"""
|
||||
Create a new admin user (super admin or platform admin).
|
||||
|
||||
Super admin only.
|
||||
"""
|
||||
# Validate platform_ids required for non-super admin
|
||||
if not request.is_super_admin and not request.platform_ids:
|
||||
raise ValidationException(
|
||||
"Platform admins must be assigned to at least one platform",
|
||||
field="platform_ids",
|
||||
)
|
||||
|
||||
if request.is_super_admin:
|
||||
# Create super admin using service
|
||||
user = admin_platform_service.create_super_admin(
|
||||
db=db,
|
||||
email=request.email,
|
||||
username=request.username,
|
||||
password=request.password,
|
||||
created_by_user_id=current_admin.id,
|
||||
first_name=request.first_name,
|
||||
last_name=request.last_name,
|
||||
)
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
|
||||
return AdminUserResponse(
|
||||
id=user.id,
|
||||
email=user.email,
|
||||
username=user.username,
|
||||
first_name=user.first_name,
|
||||
last_name=user.last_name,
|
||||
is_active=user.is_active,
|
||||
is_super_admin=user.is_super_admin,
|
||||
platform_assignments=[],
|
||||
)
|
||||
else:
|
||||
# Create platform admin with assignments using service
|
||||
user, assignments = admin_platform_service.create_platform_admin(
|
||||
db=db,
|
||||
email=request.email,
|
||||
username=request.username,
|
||||
password=request.password,
|
||||
platform_ids=request.platform_ids,
|
||||
created_by_user_id=current_admin.id,
|
||||
first_name=request.first_name,
|
||||
last_name=request.last_name,
|
||||
)
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
|
||||
return _build_admin_response(user)
|
||||
|
||||
|
||||
@router.get("/{user_id}", response_model=AdminUserResponse)
|
||||
def get_admin_user(
|
||||
user_id: int = Path(...),
|
||||
db: Session = Depends(get_db),
|
||||
current_admin: User = Depends(get_current_super_admin),
|
||||
):
|
||||
"""
|
||||
Get admin user details with platform assignments.
|
||||
|
||||
Super admin only.
|
||||
"""
|
||||
admin = admin_platform_service.get_admin_user(db=db, user_id=user_id)
|
||||
return _build_admin_response(admin)
|
||||
|
||||
|
||||
@router.post("/{user_id}/platforms/{platform_id}")
|
||||
def assign_admin_to_platform(
|
||||
user_id: int = Path(...),
|
||||
@@ -267,7 +241,7 @@ def assign_admin_to_platform(
|
||||
|
||||
Super admin only.
|
||||
"""
|
||||
assignment = admin_platform_service.assign_admin_to_platform(
|
||||
admin_platform_service.assign_admin_to_platform(
|
||||
db=db,
|
||||
admin_user_id=user_id,
|
||||
platform_id=platform_id,
|
||||
@@ -275,10 +249,6 @@ def assign_admin_to_platform(
|
||||
)
|
||||
db.commit()
|
||||
|
||||
logger.info(
|
||||
f"Assigned admin {user_id} to platform {platform_id} by {current_admin.username}"
|
||||
)
|
||||
|
||||
return {
|
||||
"message": "Admin assigned to platform successfully",
|
||||
"platform_id": platform_id,
|
||||
@@ -306,10 +276,6 @@ def remove_admin_from_platform(
|
||||
)
|
||||
db.commit()
|
||||
|
||||
logger.info(
|
||||
f"Removed admin {user_id} from platform {platform_id} by {current_admin.username}"
|
||||
)
|
||||
|
||||
return {
|
||||
"message": "Admin removed from platform successfully",
|
||||
"platform_id": platform_id,
|
||||
@@ -338,7 +304,6 @@ def toggle_super_admin_status(
|
||||
db.commit()
|
||||
|
||||
action = "promoted to" if request.is_super_admin else "demoted from"
|
||||
logger.info(f"Admin {user.username} {action} super admin by {current_admin.username}")
|
||||
|
||||
return {
|
||||
"message": f"Admin {action} super admin successfully",
|
||||
@@ -371,3 +336,54 @@ def get_admin_platforms(
|
||||
],
|
||||
"user_id": user_id,
|
||||
}
|
||||
|
||||
|
||||
@router.put("/{user_id}/status")
|
||||
def toggle_admin_status(
|
||||
user_id: int = Path(...),
|
||||
db: Session = Depends(get_db),
|
||||
current_admin: User = Depends(get_current_super_admin_api),
|
||||
):
|
||||
"""
|
||||
Toggle admin user active status.
|
||||
|
||||
Super admin only. Cannot deactivate yourself.
|
||||
"""
|
||||
admin = admin_platform_service.toggle_admin_status(
|
||||
db=db,
|
||||
user_id=user_id,
|
||||
current_admin_id=current_admin.id,
|
||||
)
|
||||
db.commit()
|
||||
|
||||
action = "activated" if admin.is_active else "deactivated"
|
||||
|
||||
return {
|
||||
"message": f"Admin user {action} successfully",
|
||||
"user_id": user_id,
|
||||
"is_active": admin.is_active,
|
||||
}
|
||||
|
||||
|
||||
@router.delete("/{user_id}")
|
||||
def delete_admin_user(
|
||||
user_id: int = Path(...),
|
||||
db: Session = Depends(get_db),
|
||||
current_admin: User = Depends(get_current_super_admin_api),
|
||||
):
|
||||
"""
|
||||
Delete an admin user.
|
||||
|
||||
Super admin only. Cannot delete yourself.
|
||||
"""
|
||||
admin_platform_service.delete_admin_user(
|
||||
db=db,
|
||||
user_id=user_id,
|
||||
current_admin_id=current_admin.id,
|
||||
)
|
||||
db.commit()
|
||||
|
||||
return {
|
||||
"message": "Admin user deleted successfully",
|
||||
"user_id": user_id,
|
||||
}
|
||||
|
||||
@@ -141,7 +141,7 @@ def get_accessible_platforms(
|
||||
- For platform admins: Only assigned platforms
|
||||
"""
|
||||
if current_user.is_super_admin:
|
||||
platforms = db.query(Platform).filter(Platform.is_active == True).all()
|
||||
platforms = admin_platform_service.get_all_active_platforms(db)
|
||||
else:
|
||||
platforms = admin_platform_service.get_platforms_for_admin(db, current_user.id)
|
||||
|
||||
@@ -184,14 +184,11 @@ def select_platform(
|
||||
"Super admins don't need platform selection - they have global access"
|
||||
)
|
||||
|
||||
# Verify admin has access to this platform
|
||||
if not current_user.can_access_platform(platform_id):
|
||||
raise InsufficientPermissionsException(
|
||||
f"You don't have access to this platform"
|
||||
)
|
||||
# Verify admin has access to this platform (raises exception if not)
|
||||
admin_platform_service.validate_admin_platform_access(current_user, platform_id)
|
||||
|
||||
# Load platform
|
||||
platform = db.query(Platform).filter(Platform.id == platform_id).first()
|
||||
platform = admin_platform_service.get_platform_by_id(db, platform_id)
|
||||
if not platform:
|
||||
raise InvalidCredentialsException("Platform not found")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user