Files
orion/app/modules/tenancy/routes/api/admin_users.py
Samir Boulahtit 9bceeaac9c feat(arch): implement soft delete for business-critical models
Adds SoftDeleteMixin (deleted_at + deleted_by_id) with automatic query
filtering via do_orm_execute event. Soft-deleted records are invisible
by default; bypass with execution_options={"include_deleted": True}.

Models: User, Merchant, Store, StoreUser, Customer, Order, Product,
LoyaltyProgram, LoyaltyCard.

Infrastructure:
- SoftDeleteMixin in models/database/base.py
- Auto query filter registered on SessionLocal and test sessions
- soft_delete(), restore(), soft_delete_cascade() in app/core/soft_delete.py
- Alembic migration adding columns to 9 tables
- Partial unique indexes on users.email/username, stores.store_code/subdomain

Service changes:
- admin_service: delete_user, delete_store → soft_delete/soft_delete_cascade
- merchant_service: delete_merchant → soft_delete_cascade (stores→children)
- store_team_service: remove_team_member → soft_delete (fixes is_active bug)
- product_service: delete_product → soft_delete
- program_service: delete_program → soft_delete_cascade

Admin API:
- include_deleted/only_deleted query params on admin list endpoints
- PUT restore endpoints for users, merchants, stores

Tests: 9 unit tests for soft-delete infrastructure.
Docs: docs/backend/soft-delete.md + follow-up proposals.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-28 21:08:07 +01:00

425 lines
12 KiB
Python

# 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,
}