# app/api/v1/admin/admin_users.py """ Admin user management endpoints (Super Admin only). This module provides endpoints for: - Listing all admin users with their platform assignments - Creating platform admins - Assigning/removing platform access - Promoting/demoting super admin status """ import logging from typing import Optional from fastapi import APIRouter, Body, Depends, Path, Query from pydantic import BaseModel, EmailStr 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.services.admin_platform_service import admin_platform_service from models.database.user import User router = APIRouter(prefix="/admin-users") logger = logging.getLogger(__name__) # ============================================================================ # SCHEMAS # ============================================================================ class PlatformAssignmentResponse(BaseModel): """Response for a platform assignment.""" platform_id: int platform_code: str platform_name: str is_active: bool class Config: from_attributes = True class AdminUserResponse(BaseModel): """Response for an admin user.""" id: int email: str username: str first_name: Optional[str] = None last_name: Optional[str] = None is_active: bool is_super_admin: bool platform_assignments: list[PlatformAssignmentResponse] = [] class Config: from_attributes = True class AdminUserListResponse(BaseModel): """Response for listing admin users.""" admins: list[AdminUserResponse] total: int class CreatePlatformAdminRequest(BaseModel): """Request to create a new platform admin.""" email: EmailStr username: str password: str first_name: Optional[str] = None last_name: Optional[str] = None platform_ids: list[int] class AssignPlatformRequest(BaseModel): """Request to assign admin to platform.""" platform_id: int class ToggleSuperAdminRequest(BaseModel): """Request to toggle super admin status.""" is_super_admin: bool # ============================================================================ # 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. """ 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") 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, ) ) return 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, ) @router.post("/{user_id}/platforms/{platform_id}") def assign_admin_to_platform( user_id: int = Path(...), platform_id: int = Path(...), db: Session = Depends(get_db), current_admin: User = Depends(get_current_super_admin_api), ): """ Assign an admin to a platform. Super admin only. """ assignment = admin_platform_service.assign_admin_to_platform( db=db, admin_user_id=user_id, platform_id=platform_id, assigned_by_user_id=current_admin.id, ) 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, "user_id": user_id, } @router.delete("/{user_id}/platforms/{platform_id}") def remove_admin_from_platform( user_id: int = Path(...), platform_id: int = Path(...), db: Session = Depends(get_db), current_admin: User = Depends(get_current_super_admin_api), ): """ Remove an admin's access to a platform. Super admin only. """ admin_platform_service.remove_admin_from_platform( db=db, admin_user_id=user_id, platform_id=platform_id, removed_by_user_id=current_admin.id, ) 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, "user_id": user_id, } @router.put("/{user_id}/super-admin") def toggle_super_admin_status( user_id: int = Path(...), request: ToggleSuperAdminRequest = Body(...), db: Session = Depends(get_db), current_admin: User = Depends(get_current_super_admin_api), ): """ Promote or demote an admin to/from super admin. Super admin only. """ user = admin_platform_service.toggle_super_admin( db=db, user_id=user_id, is_super_admin=request.is_super_admin, current_admin_id=current_admin.id, ) 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", "user_id": user_id, "is_super_admin": user.is_super_admin, } @router.get("/{user_id}/platforms") def get_admin_platforms( user_id: int = Path(...), db: Session = Depends(get_db), current_admin: User = Depends(get_current_super_admin), ): """ Get all platforms assigned to an admin. Super admin only. """ platforms = admin_platform_service.get_platforms_for_admin(db, user_id) return { "platforms": [ { "id": p.id, "code": p.code, "name": p.name, } for p in platforms ], "user_id": user_id, }