Samir Boulahtit
cad862f469
Replace direct User database model imports in API endpoints with UserContext schema, following the architecture principle that API routes should not import database models directly. Changes: - Create UserContext schema in models/schema/auth.py with from_user() factory - Update app/api/deps.py to return UserContext from all auth dependencies - Add _get_user_model() helper for functions needing User model access - Update 58 API endpoint files to use UserContext instead of User - Add noqa comments for 4 legitimate edge cases (enums, internal helpers) Architecture validation: 0 errors (down from 61), 11 warnings remain Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
398 lines
11 KiB
Python
398 lines
11 KiB
Python
# 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 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 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.exceptions import ValidationException
|
|
from app.services.admin_platform_service import admin_platform_service
|
|
from models.database.user import User # noqa: API-007 - Internal helper uses User model
|
|
from models.schema.auth import UserContext
|
|
|
|
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] = []
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
last_login: Optional[datetime] = 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: Optional[str] = None
|
|
last_name: Optional[str] = None
|
|
is_super_admin: bool = False
|
|
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
|
|
|
|
|
|
# ============================================================================
|
|
# 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,
|
|
is_super_admin=admin.is_super_admin,
|
|
platform_assignments=assignments,
|
|
created_at=admin.created_at,
|
|
updated_at=admin.updated_at,
|
|
last_login=admin.last_login,
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# 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: 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,
|
|
)
|
|
|
|
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: 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 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: 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)
|
|
|
|
|
|
@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,
|
|
}
|
|
|
|
|
|
@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,
|
|
}
|
|
|
|
|
|
@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.is_super_admin,
|
|
current_admin_id=current_admin.id,
|
|
)
|
|
db.commit()
|
|
|
|
action = "promoted to" if request.is_super_admin else "demoted from"
|
|
|
|
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: 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,
|
|
}
|
|
|
|
|
|
@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,
|
|
}
|
|
|
|
|
|
@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,
|
|
}
|