46 lines
1.2 KiB
Python
46 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter
|
|
from fastapi.responses import JSONResponse
|
|
from sqlalchemy import text
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.database import SessionLocal
|
|
from app.core.responses import api_success
|
|
from app.services.redis_client import get_sync_redis_client
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/health")
|
|
def health() -> dict:
|
|
return api_success({"status": "ok"})
|
|
|
|
|
|
@router.get("/ready")
|
|
def ready():
|
|
checks = {"database": _database_ready(), "redis": _redis_ready()}
|
|
redis_required = get_settings().readiness_requires_redis
|
|
ready_state = checks["database"] and (checks["redis"] or not redis_required)
|
|
payload = api_success({"status": "ready" if ready_state else "not_ready", "checks": checks})
|
|
return payload if ready_state else JSONResponse(status_code=503, content=payload)
|
|
|
|
|
|
def _database_ready() -> bool:
|
|
try:
|
|
with SessionLocal() as db:
|
|
db.execute(text("SELECT 1"))
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _redis_ready() -> bool:
|
|
client = get_sync_redis_client()
|
|
if client is None:
|
|
return False
|
|
try:
|
|
return bool(client.ping())
|
|
except Exception:
|
|
return False
|