xiaoxia-saas/apps/api/app/api/routes/health.py

from datetime import datetime, timezone

import psycopg2
import redis
from app.config import settings
from fastapi import APIRouter, status
from fastapi.responses import JSONResponse

router = APIRouter(tags=["Health"])


@router.get("/health", status_code=status.HTTP_200_OK)
async def health_check():
    return {
        "status": "healthy",
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "version": settings.APP_VERSION,
    }


@router.get("/ready", status_code=status.HTTP_200_OK)
async def readiness_check():
    """简单的就绪检查，仅返回状态。详细健康检查请使用 /health 端点。"""
    return {"status": "ready"}


@router.get("/startup", status_code=status.HTTP_200_OK)
async def startup_check():
    checks = {
        "database": await _check_database(),
        "migrations": await _check_migrations(),
    }
    all_ready = all(check["status"] == "healthy" for check in checks.values())
    response = {
        "status": "started" if all_ready else "starting",
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "checks": checks,
    }
    if not all_ready:
        return JSONResponse(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, content=response)
    return response


async def _check_database() -> dict:
    if settings.USE_IN_MEMORY_DB:
        return {
            "status": "healthy",
            "type": "in_memory",
            "message": "Using in-memory database",
        }
    try:
        conn = psycopg2.connect(settings.DATABASE_URL, connect_timeout=3)
        with conn.cursor() as cur:
            cur.execute("SELECT 1")
            cur.fetchone()
        conn.close()
        return {
            "status": "healthy",
            "type": "postgresql",
            "message": "Database connection successful",
        }
    except Exception as error:
        return {
            "status": "unhealthy",
            "type": "postgresql",
            "message": f"Database connection failed: {error}",
        }


async def _check_redis() -> dict:
    try:
        client = redis.from_url(settings.REDIS_URL, socket_connect_timeout=3)
        client.ping()
        client.close()
        return {
            "status": "healthy",
            "type": "redis",
            "message": "Redis connection successful",
        }
    except Exception as error:
        return {
            "status": "unhealthy",
            "type": "redis",
            "message": f"Redis connection failed: {error}",
        }


def _check_oss() -> dict:
    try:
        from app.core.storage import get_storage_service

        svc = get_storage_service()
        if not svc.access_key_id or not svc.access_key_secret:
            return {
                "status": "unhealthy",
                "type": "oss",
                "message": "OSS credentials not configured (OSS_ACCESS_KEY_ID / OSS_ACCESS_KEY_SECRET missing)",
            }
        if svc.bucket is None:
            return {
                "status": "unhealthy",
                "type": "oss",
                "message": "OSS SDK (oss2) not installed or bucket client init failed",
            }
        # Try a lightweight OSS API call to verify connectivity & credentials
        svc.bucket.get_bucket_info()
        return {
            "status": "healthy",
            "type": "oss",
            "message": f"OSS connected: endpoint={svc.endpoint} bucket={svc.bucket_name}",
        }
    except Exception as error:
        return {
            "status": "unhealthy",
            "type": "oss",
            "message": f"OSS check failed: {type(error).__name__}: {error}",
        }


async def _check_migrations() -> dict:
    if settings.USE_IN_MEMORY_DB:
        return {
            "status": "healthy",
            "message": "Using in-memory database, no migrations needed",
        }
    try:
        conn = psycopg2.connect(settings.DATABASE_URL, connect_timeout=3)
        with conn.cursor() as cur:
            cur.execute("""
                SELECT COUNT(*) FROM information_schema.tables
                WHERE table_name IN ('projects', 'asset_libraries', 'assets', 'ingest_jobs', 'classification_jobs')
                """)
            count = cur.fetchone()[0]
        conn.close()
        if count >= 5:
            return {"status": "healthy", "message": "Database migrations applied"}
        return {"status": "unhealthy", "message": f"Missing tables, found {count}/5"}
    except Exception as error:
        return {"status": "unhealthy", "message": f"Migration check failed: {error}"}