152 lines
4.8 KiB
Python
152 lines
4.8 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
|
|
from app.admin_auth import require_admin, require_super_admin
|
|
from app.admin_models import (
|
|
DeleteUserResponse,
|
|
HardResetResponse,
|
|
InvariantsResponse,
|
|
SupportTicketsResponse,
|
|
DeleteSupportTicketResponse,
|
|
OverviewResponse,
|
|
RunsResponse,
|
|
RunDetailResponse,
|
|
UsersResponse,
|
|
UserDetailResponse,
|
|
)
|
|
from app.admin_service import (
|
|
delete_user_hard,
|
|
hard_reset_user_data,
|
|
get_invariants,
|
|
get_support_tickets,
|
|
delete_support_ticket,
|
|
get_overview,
|
|
get_run_detail,
|
|
get_runs,
|
|
get_user_detail,
|
|
get_users,
|
|
)
|
|
from app.admin_role_service import set_user_role
|
|
|
|
router = APIRouter(prefix="/api/admin", dependencies=[Depends(require_admin)])
|
|
|
|
|
|
@router.get("/overview", response_model=OverviewResponse)
|
|
def admin_overview():
|
|
return get_overview()
|
|
|
|
|
|
@router.get("/users", response_model=UsersResponse)
|
|
def admin_users(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=200),
|
|
query: str | None = None,
|
|
):
|
|
return get_users(page, page_size, query)
|
|
|
|
|
|
@router.get("/users/{user_id}", response_model=UserDetailResponse)
|
|
def admin_user_detail(user_id: str):
|
|
detail = get_user_detail(user_id)
|
|
if not detail:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return detail
|
|
|
|
|
|
@router.delete("/users/{user_id}", response_model=DeleteUserResponse)
|
|
def admin_delete_user(
|
|
user_id: str,
|
|
hard: bool = Query(False),
|
|
admin_user: dict = Depends(require_super_admin),
|
|
):
|
|
if not hard:
|
|
raise HTTPException(status_code=400, detail="Hard delete requires hard=true")
|
|
result = delete_user_hard(user_id, admin_user)
|
|
if result is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return result
|
|
|
|
|
|
@router.post("/users/{user_id}/hard-reset", response_model=HardResetResponse)
|
|
def admin_hard_reset_user(
|
|
user_id: str,
|
|
admin_user: dict = Depends(require_super_admin),
|
|
):
|
|
result = hard_reset_user_data(user_id, admin_user)
|
|
if result is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return result
|
|
|
|
|
|
@router.post("/users/{user_id}/make-admin")
|
|
def admin_make_admin(user_id: str, admin_user: dict = Depends(require_super_admin)):
|
|
result = set_user_role(admin_user["id"], user_id, "ADMIN")
|
|
if result is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
if result.get("error") == "cannot_demote_self":
|
|
raise HTTPException(status_code=400, detail="Cannot demote self")
|
|
if result.get("error") == "invalid_role":
|
|
raise HTTPException(status_code=400, detail="Invalid role")
|
|
return result
|
|
|
|
|
|
@router.post("/users/{user_id}/revoke-admin")
|
|
def admin_revoke_admin(user_id: str, admin_user: dict = Depends(require_super_admin)):
|
|
result = set_user_role(admin_user["id"], user_id, "USER")
|
|
if result is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
if result.get("error") == "cannot_demote_self":
|
|
raise HTTPException(status_code=400, detail="Cannot demote self")
|
|
if result.get("error") == "invalid_role":
|
|
raise HTTPException(status_code=400, detail="Invalid role")
|
|
return result
|
|
|
|
|
|
@router.post("/users/{user_id}/make-super-admin")
|
|
def admin_make_super_admin(user_id: str, admin_user: dict = Depends(require_super_admin)):
|
|
result = set_user_role(admin_user["id"], user_id, "SUPER_ADMIN")
|
|
if result is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
if result.get("error") == "invalid_role":
|
|
raise HTTPException(status_code=400, detail="Invalid role")
|
|
return result
|
|
|
|
|
|
@router.get("/runs", response_model=RunsResponse)
|
|
def admin_runs(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=200),
|
|
status: str | None = None,
|
|
mode: str | None = None,
|
|
user_id: str | None = None,
|
|
):
|
|
return get_runs(page, page_size, status, mode, user_id)
|
|
|
|
|
|
@router.get("/runs/{run_id}", response_model=RunDetailResponse)
|
|
def admin_run_detail(run_id: str):
|
|
detail = get_run_detail(run_id)
|
|
if not detail:
|
|
raise HTTPException(status_code=404, detail="Run not found")
|
|
return detail
|
|
|
|
|
|
@router.get("/health/invariants", response_model=InvariantsResponse)
|
|
def admin_invariants():
|
|
return get_invariants()
|
|
|
|
|
|
@router.get("/support-tickets", response_model=SupportTicketsResponse)
|
|
def admin_support_tickets(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=200),
|
|
):
|
|
return get_support_tickets(page, page_size)
|
|
|
|
|
|
@router.delete("/support-tickets/{ticket_id}", response_model=DeleteSupportTicketResponse)
|
|
def admin_delete_support_ticket(ticket_id: str):
|
|
result = delete_support_ticket(ticket_id)
|
|
if not result:
|
|
raise HTTPException(status_code=404, detail="Ticket not found")
|
|
return result
|