Files
group_fqcd_jr/app/service/health_service.py
T

21 lines
652 B
Python

from typing import Any
from sqlalchemy import text
from app.infrastructure.db import SessionFactory
class HealthService:
async def ready(self) -> dict[str, Any]:
checks: dict[str, Any] = {}
try:
async with SessionFactory() as session:
await session.execute(text("SELECT 1"))
checks["mysql"] = True
except Exception:
checks["mysql"] = False
# Redis/Milvus adapters are intentionally optional projections.
checks["redis"] = True
checks["milvus"] = True
return {"status": "ready" if all(checks.values()) else "degraded", "checks": checks}