46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
from fastapi import APIRouter
|
|
from fastapi.responses import JSONResponse
|
|
from sqlalchemy import text
|
|
|
|
from alembic.config import Config
|
|
from alembic.runtime.migration import MigrationContext
|
|
from alembic.script import ScriptDirectory
|
|
|
|
from app.persistence.db import engine
|
|
from app.settings import get_settings
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _alembic_at_head() -> bool:
|
|
config = Config("alembic.ini")
|
|
script = ScriptDirectory.from_config(config)
|
|
with engine.connect() as connection:
|
|
context = MigrationContext.configure(connection)
|
|
return set(context.get_current_heads()) == set(script.get_heads())
|
|
|
|
|
|
@router.get("/health/live")
|
|
def live() -> dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.get("/health/ready", response_model=None)
|
|
def ready() -> JSONResponse | dict[str, str]:
|
|
settings = get_settings()
|
|
try:
|
|
with engine.connect() as connection:
|
|
connection.execute(text("SELECT 1"))
|
|
if not _alembic_at_head():
|
|
return JSONResponse(
|
|
{"status": "not_ready", "reason": "migration_behind_head"},
|
|
status_code=503,
|
|
)
|
|
except Exception:
|
|
return JSONResponse({"status": "not_ready", "reason": "database"}, status_code=503)
|
|
return {
|
|
"status": "ready",
|
|
"environment": settings.environment,
|
|
"engine_build": settings.engine_build,
|
|
}
|