Implementation Engineer
26 lines
748 B
Python
26 lines
748 B
Python
"""Health check router."""
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from app.database import get_db, engine
|
|
from app.cache import cache
|
|
|
|
router = APIRouter(prefix="/api/health", tags=["health"])
|
|
|
|
|
|
@router.get("")
|
|
async def health_check(db: AsyncSession = Depends(get_db)) -> dict:
|
|
"""Return health status: app, db, redis."""
|
|
db_ok = "connected"
|
|
redis_ok = "connected"
|
|
try:
|
|
await db.execute(text("SELECT 1"))
|
|
except Exception:
|
|
db_ok = "disconnected"
|
|
try:
|
|
await cache.connect()
|
|
await cache._redis.ping()
|
|
except Exception:
|
|
redis_ok = "disconnected"
|
|
return {"status": "ok", "db": db_ok, "redis": redis_ok}
|