102 lines
2.9 KiB
Python
102 lines
2.9 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
SysAdmin API for database table statistics and FK orphan detection/cleanup.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from pydantic import BaseModel, Field
|
|
|
|
from modules.auth import limiter
|
|
from modules.auth.authentication import requireSysAdminRole
|
|
from modules.datamodels.datamodelUam import User
|
|
from modules.system.databaseHealth import (
|
|
_cleanAllOrphans,
|
|
_cleanOrphans,
|
|
_getTableStats,
|
|
_scanOrphans,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(
|
|
prefix="/api/admin/database-health",
|
|
tags=["Admin Database Health"],
|
|
)
|
|
|
|
|
|
class OrphanCleanRequest(BaseModel):
|
|
"""Body for deleting orphans for one FK relationship."""
|
|
|
|
db: str = Field(..., description="Source database name (e.g. poweron_app)")
|
|
table: str = Field(..., description="Source table (Pydantic model class name)")
|
|
column: str = Field(..., description="FK column on the source table")
|
|
|
|
|
|
@router.get("/stats")
|
|
@limiter.limit("30/minute")
|
|
def getDatabaseTableStats(
|
|
request: Request,
|
|
db: Optional[str] = None,
|
|
currentUser: User = Depends(requireSysAdminRole),
|
|
) -> Dict[str, Any]:
|
|
"""Table statistics from pg_stat_user_tables (optional filter by database name)."""
|
|
rows = _getTableStats(dbFilter=db)
|
|
return {"stats": rows}
|
|
|
|
|
|
@router.get("/orphans")
|
|
@limiter.limit("10/minute")
|
|
def getDatabaseOrphans(
|
|
request: Request,
|
|
db: Optional[str] = None,
|
|
currentUser: User = Depends(requireSysAdminRole),
|
|
) -> Dict[str, Any]:
|
|
"""FK orphan scan (optional filter by source database name)."""
|
|
rows = _scanOrphans(dbFilter=db)
|
|
return {"orphans": rows}
|
|
|
|
|
|
@router.post("/orphans/clean")
|
|
@limiter.limit("10/minute")
|
|
def postDatabaseOrphansClean(
|
|
request: Request,
|
|
body: OrphanCleanRequest,
|
|
currentUser: User = Depends(requireSysAdminRole),
|
|
) -> Dict[str, Any]:
|
|
"""Delete orphaned rows for a single FK relationship."""
|
|
try:
|
|
deleted = _cleanOrphans(body.db, body.table, body.column)
|
|
except ValueError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=str(e),
|
|
) from e
|
|
logger.info(
|
|
"SysAdmin orphan clean: user=%s db=%s table=%s column=%s deleted=%s",
|
|
currentUser.username,
|
|
body.db,
|
|
body.table,
|
|
body.column,
|
|
deleted,
|
|
)
|
|
return {"deleted": deleted}
|
|
|
|
|
|
@router.post("/orphans/clean-all")
|
|
@limiter.limit("2/minute")
|
|
def postDatabaseOrphansCleanAll(
|
|
request: Request,
|
|
currentUser: User = Depends(requireSysAdminRole),
|
|
) -> Dict[str, Any]:
|
|
"""Run orphan cleanup for every relationship that currently has orphans."""
|
|
results: List[dict] = _cleanAllOrphans()
|
|
logger.info(
|
|
"SysAdmin orphan clean-all: user=%s batches=%s",
|
|
currentUser.username,
|
|
len(results),
|
|
)
|
|
return {"results": results}
|