File size: 11,185 Bytes
6aecb2e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 | """
Database management API endpoints
"""
from typing import List, Optional, Dict, Any
from fastapi import APIRouter, HTTPException, Query, Depends
from pydantic import BaseModel
from ..auth.middleware import get_current_admin_user, get_current_user_required
from ..database.models import User
from ..database.health_check import health_checker
from ..database.migrations import migration_manager
from ..services.db_project_manager import DatabaseProjectManager
router = APIRouter(prefix="/api/database", tags=["database"])
class HealthCheckResponse(BaseModel):
"""Health check response model"""
overall_status: str
timestamp: float
critical_failures: List[str]
checks: Dict[str, Any]
summary: Dict[str, int]
class MigrationStatusResponse(BaseModel):
"""Migration status response model"""
current_version: Optional[str]
applied_migrations: List[str]
available_migrations: List[str]
pending_migrations: List[str]
class DatabaseStatsResponse(BaseModel):
"""Database statistics response model"""
status: str
stats: Dict[str, Any]
timestamp: float
@router.get("/health", response_model=HealthCheckResponse)
async def get_database_health(
checks: Optional[List[str]] = Query(None, description="Specific checks to run"),
_admin_user: User = Depends(get_current_admin_user),
):
"""
Get database health status
Available checks:
- connection: Database connection test
- tables: Table existence and structure
- data_integrity: Data integrity and relationships
- performance: Database performance metrics
- storage: Storage usage and optimization
"""
try:
result = await health_checker.run_health_check(checks)
return HealthCheckResponse(**result)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Health check failed: {str(e)}")
@router.get("/health/quick")
async def get_quick_health_check(
_admin_user: User = Depends(get_current_admin_user),
):
"""
Quick health check (connection only)
"""
try:
result = await health_checker.run_health_check(["connection"])
return {
"status": result["overall_status"],
"timestamp": result["timestamp"],
"connection": result["checks"]["connection"]["status"]
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Quick health check failed: {str(e)}")
@router.get("/stats", response_model=DatabaseStatsResponse)
async def get_database_stats(
_admin_user: User = Depends(get_current_admin_user),
):
"""
Get comprehensive database statistics
"""
try:
result = await health_checker.get_database_stats()
return DatabaseStatsResponse(**result)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to get database stats: {str(e)}")
@router.get("/migrations/status", response_model=MigrationStatusResponse)
async def get_migration_status(
_admin_user: User = Depends(get_current_admin_user),
):
"""
Get database migration status
"""
try:
result = await migration_manager.get_migration_status()
if "error" in result:
raise HTTPException(status_code=500, detail=result["error"])
return MigrationStatusResponse(**result)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to get migration status: {str(e)}")
@router.post("/migrations/run")
async def run_migrations(
target_version: Optional[str] = None,
rollback: bool = False,
_admin_user: User = Depends(get_current_admin_user),
):
"""
Run database migrations
- target_version: Target migration version (optional)
- rollback: Whether to rollback migrations (requires target_version)
"""
try:
if rollback:
if not target_version:
raise HTTPException(status_code=400, detail="Target version required for rollback")
success = await migration_manager.migrate_down(target_version)
action = "rollback"
else:
success = await migration_manager.migrate_up(target_version)
action = "migration"
if success:
return {
"status": "success",
"message": f"Database {action} completed successfully",
"target_version": target_version
}
else:
raise HTTPException(status_code=500, detail=f"Database {action} failed")
except Exception as e:
raise HTTPException(status_code=500, detail=f"Migration operation failed: {str(e)}")
@router.get("/projects/summary")
async def get_projects_summary(
user: User = Depends(get_current_user_required),
):
"""
Get summary of projects in database
"""
try:
project_manager = DatabaseProjectManager()
# Get project list
scoped_user_id = None if user.is_admin else user.id
project_list = await project_manager.list_projects(page=1, page_size=100, user_id=scoped_user_id)
# Calculate summary statistics
total_projects = project_list.total
status_counts = {}
scenario_counts = {}
for project in project_list.projects:
# Count by status
status = project.status
status_counts[status] = status_counts.get(status, 0) + 1
# Count by scenario
scenario = project.scenario
scenario_counts[scenario] = scenario_counts.get(scenario, 0) + 1
await project_manager.close()
return {
"total_projects": total_projects,
"status_distribution": status_counts,
"scenario_distribution": scenario_counts,
"recent_projects": len([p for p in project_list.projects[:10]]) # Last 10 projects
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to get projects summary: {str(e)}")
@router.delete("/projects/{project_id}")
async def delete_project(
project_id: str,
user: User = Depends(get_current_user_required),
):
"""
Delete a project from database
"""
try:
project_manager = DatabaseProjectManager()
# Check if project exists
scoped_user_id = None if user.is_admin else user.id
project = await project_manager.get_project(project_id, user_id=scoped_user_id)
if not project:
raise HTTPException(status_code=404, detail="Project not found")
# Delete project
success = await project_manager.delete_project(project_id, user_id=scoped_user_id)
await project_manager.close()
if success:
return {
"status": "success",
"message": f"Project {project_id} deleted successfully"
}
else:
raise HTTPException(status_code=500, detail="Failed to delete project")
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to delete project: {str(e)}")
@router.post("/cleanup/orphaned")
async def cleanup_orphaned_data(
_admin_user: User = Depends(get_current_admin_user),
):
"""
Clean up orphaned data in database
"""
try:
from ..database.database import AsyncSessionLocal
from sqlalchemy import text
cleanup_results = {}
async with AsyncSessionLocal() as session:
# Clean up orphaned todo_boards
result = await session.execute(text("""
DELETE FROM todo_boards
WHERE project_id NOT IN (SELECT project_id FROM projects)
"""))
cleanup_results["orphaned_todo_boards"] = result.rowcount
# Clean up orphaned todo_stages
result = await session.execute(text("""
DELETE FROM todo_stages
WHERE todo_board_id NOT IN (SELECT id FROM todo_boards)
"""))
cleanup_results["orphaned_todo_stages"] = result.rowcount
# Clean up orphaned slide_data
result = await session.execute(text("""
DELETE FROM slide_data
WHERE project_id NOT IN (SELECT project_id FROM projects)
"""))
cleanup_results["orphaned_slide_data"] = result.rowcount
# Clean up orphaned project_versions
result = await session.execute(text("""
DELETE FROM project_versions
WHERE project_id NOT IN (SELECT project_id FROM projects)
"""))
cleanup_results["orphaned_project_versions"] = result.rowcount
await session.commit()
total_cleaned = sum(cleanup_results.values())
return {
"status": "success",
"message": f"Cleaned up {total_cleaned} orphaned records",
"details": cleanup_results
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Cleanup failed: {str(e)}")
@router.get("/backup/info")
async def get_backup_info(
_admin_user: User = Depends(get_current_admin_user),
):
"""
Get backup information and recommendations
"""
try:
import os
from ..core.config import app_config
# Get database file info
db_url = app_config.database_url
if db_url.startswith("sqlite:///"):
db_file = db_url.replace("sqlite:///", "")
if db_file.startswith("./"):
db_file = db_file[2:]
if os.path.exists(db_file):
file_size = os.path.getsize(db_file)
file_modified = os.path.getmtime(db_file)
return {
"database_type": "SQLite",
"database_file": db_file,
"file_size_bytes": file_size,
"file_size_mb": round(file_size / (1024 * 1024), 2),
"last_modified": file_modified,
"backup_recommendation": "Use 'python manage_database.py backup <filename>' to create backup",
"restore_recommendation": "Use 'python manage_database.py restore <filename>' to restore backup"
}
else:
return {
"database_type": "SQLite",
"database_file": db_file,
"status": "not_found",
"message": "Database file not found"
}
else:
return {
"database_type": "Other",
"database_url": db_url,
"backup_recommendation": "Use database-specific backup tools",
"message": "Backup info only available for SQLite databases"
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to get backup info: {str(e)}")
|