INNER CODE UNIT · Python

_check_db_connectivity_async

KatrielMoses/voidaccess · api/main.py:518

async def _check_db_connectivity_async() -> str:
    """Return 'ok' if DB is reachable, error message otherwise."""
    if not os.getenv("DATABASE_URL"):
        return "error: DATABASE_URL not configured"
    try:
        from db.session import get_async_session  # noqa: PLC0415
        from sqlalchemy import text

        async with get_async_session() as session:
            await session.execute(text("SELECT 1"))
        return "ok"
    except Exception as exc:
        # /health and /healthz/ready are unauthenticated — never echo the raw
        # DB error (can contain connection-string fragments). Log it, return
        # a generic status.
        log_exception(exc, context="db connectivity check")
        return "error: database unreachable"

View source record →