# app/modules/tenancy/routes/api/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 and super admins - Assigning/removing platform access - Promoting/demoting super admin status - Toggling admin status - Deleting admin users """ import logging from datetime import datetime 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.exceptions import ValidationException from app.modules.tenancy.models import ( User, # API-007 - Internal helper uses User model ) from app.modules.tenancy.schemas.auth import UserContext from app.modules.tenancy.services.admin_platform_service import admin_platform_service admin_users_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: str | None = None last_name: str | None = None is_active: bool role: str platform_assignments: list[PlatformAssignmentResponse] = [] created_at: datetime updated_at: datetime last_login: datetime | None = None class Config: from_attributes = True class AdminUserListResponse(BaseModel): """Response for listing admin users.""" admins: list[AdminUserResponse] total: int class CreateAdminUserRequest(BaseModel): """Request to create a new admin user (platform admin or super admin).""" email: EmailStr username: str password: str first_name: str | None = None last_name: str | None = None role: str = "platform_admin" platform_ids: list[int] = [] class AssignPlatformRequest(BaseModel): """Request to assign admin to platform.""" platform_id: int class ToggleSuperAdminRequest(BaseModel): """Request to change admin role (super_admin or platform_admin).""" role: str # "super_admin" or "platform_admin" # ============================================================================ # HELPER FUNCTIONS # ============================================================================ 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: 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, role=admin.role, platform_assignments=assignments, created_at=admin.created_at, updated_at=admin.updated_at, last_login=admin.last_login, ) # ============================================================================ # ENDPOINTS # ============================================================================ @admin_users_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), include_deleted: bool = Query(False, description="Include soft-deleted users"), only_deleted: bool = Query(False, description="Show only soft-deleted users (trash view)"), db: Session = Depends(get_db), current_admin: UserContext = 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, include_deleted=include_deleted, only_deleted=only_deleted, ) admin_responses = [_build_admin_response(admin) for admin in admins] return AdminUserListResponse(admins=admin_responses, total=total) @admin_users_router.post("", response_model=AdminUserResponse) def create_admin_user( request: CreateAdminUserRequest, db: Session = Depends(get_db), current_admin: UserContext = Depends(get_current_super_admin_api), ): """ Create a new admin user (super admin or platform admin). Super admin only. """ # Validate platform_ids required for platform admins if request.role != "super_admin" and not request.platform_ids: raise ValidationException( "Platform admins must be assigned to at least one platform", field="platform_ids", ) if request.role == "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, role=user.role, platform_assignments=[], ) # 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) @admin_users_router.get("/{user_id}", response_model=AdminUserResponse) def get_admin_user( user_id: int = Path(...), db: Session = Depends(get_db), current_admin: UserContext = 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) @admin_users_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: UserContext = Depends(get_current_super_admin_api), ): """ Assign an admin to a platform. Super admin only. """ 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() return { "message": "Admin assigned to platform successfully", "platform_id": platform_id, "user_id": user_id, } @admin_users_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: UserContext = 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() return { "message": "Admin removed from platform successfully", "platform_id": platform_id, "user_id": user_id, } @admin_users_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: UserContext = 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.role == "super_admin"), current_admin_id=current_admin.id, ) db.commit() action = "promoted to" if request.role == "super_admin" else "demoted from" return { "message": f"Admin {action} super admin successfully", "user_id": user_id, "role": user.role, } @admin_users_router.get("/{user_id}/platforms") def get_admin_platforms( user_id: int = Path(...), db: Session = Depends(get_db), current_admin: UserContext = 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, } @admin_users_router.put("/{user_id}/status") def toggle_admin_status( user_id: int = Path(...), db: Session = Depends(get_db), current_admin: UserContext = 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, } @admin_users_router.delete("/{user_id}") def delete_admin_user( user_id: int = Path(...), db: Session = Depends(get_db), current_admin: UserContext = 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, } @admin_users_router.put("/{user_id}/restore") def restore_admin_user( user_id: int = Path(...), db: Session = Depends(get_db), current_admin: UserContext = Depends(get_current_super_admin_api), ): """ Restore a soft-deleted admin user. Super admin only. """ from app.core.soft_delete import restore from app.modules.tenancy.models import User restored = restore(db, User, user_id, restored_by_id=current_admin.id) db.commit() return { "message": f"User '{restored.username}' restored successfully", "user_id": user_id, }