Samir Boulahtit
b98c7c553b
- images.py: delegate file validation (size, type, extension) to image_service - platform_health.py: extract all database queries to platform_health_service - Fixes 13 architecture validation errors (API-002, API-003) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
147 lines
3.9 KiB
Python
147 lines
3.9 KiB
Python
# app/api/v1/admin/platform_health.py
|
|
"""
|
|
Platform health and capacity monitoring endpoints.
|
|
|
|
Provides:
|
|
- Overall platform health status
|
|
- Capacity metrics and thresholds
|
|
- Scaling recommendations
|
|
"""
|
|
|
|
import logging
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import get_current_admin_api
|
|
from app.core.database import get_db
|
|
from app.services.platform_health_service import platform_health_service
|
|
from models.database.user import User
|
|
|
|
router = APIRouter()
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ============================================================================
|
|
# Schemas
|
|
# ============================================================================
|
|
|
|
|
|
class SystemMetrics(BaseModel):
|
|
"""System resource metrics."""
|
|
|
|
cpu_percent: float
|
|
memory_percent: float
|
|
memory_used_gb: float
|
|
memory_total_gb: float
|
|
disk_percent: float
|
|
disk_used_gb: float
|
|
disk_total_gb: float
|
|
|
|
|
|
class DatabaseMetrics(BaseModel):
|
|
"""Database metrics."""
|
|
|
|
size_mb: float
|
|
products_count: int
|
|
orders_count: int
|
|
vendors_count: int
|
|
inventory_count: int
|
|
|
|
|
|
class ImageStorageMetrics(BaseModel):
|
|
"""Image storage metrics."""
|
|
|
|
total_files: int
|
|
total_size_mb: float
|
|
total_size_gb: float
|
|
max_files_per_dir: int
|
|
products_estimated: int
|
|
|
|
|
|
class CapacityThreshold(BaseModel):
|
|
"""Capacity threshold status."""
|
|
|
|
name: str
|
|
current: float
|
|
warning: float
|
|
critical: float
|
|
limit: float
|
|
status: str # ok, warning, critical
|
|
percent_used: float
|
|
|
|
|
|
class ScalingRecommendation(BaseModel):
|
|
"""Scaling recommendation."""
|
|
|
|
priority: str # info, warning, critical
|
|
title: str
|
|
description: str
|
|
action: str | None = None
|
|
|
|
|
|
class PlatformHealthResponse(BaseModel):
|
|
"""Complete platform health response."""
|
|
|
|
timestamp: str
|
|
overall_status: str # healthy, degraded, critical
|
|
system: SystemMetrics
|
|
database: DatabaseMetrics
|
|
image_storage: ImageStorageMetrics
|
|
thresholds: list[CapacityThreshold]
|
|
recommendations: list[ScalingRecommendation]
|
|
infrastructure_tier: str
|
|
next_tier_trigger: str | None = None
|
|
|
|
|
|
class CapacityMetricsResponse(BaseModel):
|
|
"""Capacity-focused metrics."""
|
|
|
|
products_total: int
|
|
products_by_vendor: dict[str, int]
|
|
images_total: int
|
|
storage_used_gb: float
|
|
database_size_mb: float
|
|
orders_this_month: int
|
|
active_vendors: int
|
|
|
|
|
|
# ============================================================================
|
|
# Endpoints
|
|
# ============================================================================
|
|
|
|
|
|
@router.get("/health", response_model=PlatformHealthResponse)
|
|
async def get_platform_health(
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_api),
|
|
):
|
|
"""Get comprehensive platform health status.
|
|
|
|
Returns system metrics, database stats, storage info, and recommendations.
|
|
"""
|
|
health_data = platform_health_service.get_full_health_report(db)
|
|
|
|
return PlatformHealthResponse(
|
|
timestamp=health_data["timestamp"],
|
|
overall_status=health_data["overall_status"],
|
|
system=SystemMetrics(**health_data["system"]),
|
|
database=DatabaseMetrics(**health_data["database"]),
|
|
image_storage=ImageStorageMetrics(**health_data["image_storage"]),
|
|
thresholds=[CapacityThreshold(**t) for t in health_data["thresholds"]],
|
|
recommendations=[ScalingRecommendation(**r) for r in health_data["recommendations"]],
|
|
infrastructure_tier=health_data["infrastructure_tier"],
|
|
next_tier_trigger=health_data["next_tier_trigger"],
|
|
)
|
|
|
|
|
|
@router.get("/capacity", response_model=CapacityMetricsResponse)
|
|
async def get_capacity_metrics(
|
|
db: Session = Depends(get_db),
|
|
current_admin: User = Depends(get_current_admin_api),
|
|
):
|
|
"""Get capacity-focused metrics for planning."""
|
|
metrics = platform_health_service.get_capacity_metrics(db)
|
|
return CapacityMetricsResponse(**metrics)
|