40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
from fastapi import APIRouter
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import OperationalError
|
|
|
|
from app.config import settings
|
|
from app.database import engine
|
|
from app.mongo import mongo_client, mongo_db
|
|
|
|
router = APIRouter(tags=["health"])
|
|
|
|
|
|
@router.get("/health")
|
|
async def health_check() -> dict[str, str]:
|
|
db_status = "disconnected"
|
|
try:
|
|
with engine.connect() as conn:
|
|
conn.execute(text("SELECT 1"))
|
|
db_status = "sqlite" if settings.is_sqlite else "connected"
|
|
except OperationalError:
|
|
pass
|
|
|
|
mongo_status = "disconnected"
|
|
if mongo_client is not None and mongo_db is not None:
|
|
try:
|
|
await mongo_client.admin.command("ping")
|
|
mongo_status = "connected"
|
|
except Exception:
|
|
mongo_status = "disconnected"
|
|
|
|
overall = "ok" if db_status != "disconnected" and mongo_status == "connected" else "degraded"
|
|
|
|
return {
|
|
"status": overall,
|
|
"service": "silos-api",
|
|
"database": db_status,
|
|
"mongodb": mongo_status,
|
|
"hint": ""
|
|
if mongo_status == "connected"
|
|
else "Start MongoDB with: docker compose up mongo -d",
|
|
}
|