44 lines
1.0 KiB
Python
44 lines
1.0 KiB
Python
from datetime import datetime, timezone
|
|
|
|
import redis as redis_lib
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.config import settings
|
|
from app.database import get_db
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("")
|
|
async def health():
|
|
return {"status": "ok", "timestamp": datetime.now(timezone.utc).isoformat()}
|
|
|
|
|
|
@router.get("/detailed")
|
|
async def health_detailed(db: AsyncSession = Depends(get_db)):
|
|
# Check DB
|
|
db_ok = False
|
|
try:
|
|
await db.execute(text("SELECT 1"))
|
|
db_ok = True
|
|
except Exception:
|
|
pass
|
|
|
|
# Check Redis
|
|
redis_ok = False
|
|
try:
|
|
r = redis_lib.from_url(settings.REDIS_URL)
|
|
redis_ok = r.ping()
|
|
except Exception:
|
|
pass
|
|
|
|
status = "ok" if (db_ok and redis_ok) else "degraded"
|
|
return {
|
|
"status": status,
|
|
"database": "ok" if db_ok else "error",
|
|
"redis": "ok" if redis_ok else "error",
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
}
|