scripts/check_db_connectivity.py

#!/usr/bin/env python3
from __future__ import annotations

import os
from sqlalchemy import create_engine, inspect, text

TABLES_TO_CHECK = [
    "otree_session",
    "otree_participant",
    "otree_page_completion",
]


def main() -> int:
    url = os.environ.get("DATABASE_URL")
    if not url:
        print("[db-debug] DATABASE_URL not set")
        return 0

    print(f"[db-debug] Connecting with URL: {url}")
    engine = create_engine(url)
    try:
        with engine.connect() as conn:
            print(f"[db-debug] Connected; dialect={conn.dialect.name}")
            inspector = inspect(conn)
            for table in TABLES_TO_CHECK:
                if not inspector.has_table(table):
                    print(f"[db-debug] {table}: missing (skipping)")
                    continue
                try:
                    count = conn.execute(text(f"SELECT COUNT(*) FROM {table}")).scalar()
                    print(f"[db-debug] {table}: {count}")
                except Exception as exc:  # noqa: BLE001
                    print(f"[db-debug] {table}: error {exc}")
    except Exception as exc:  # noqa: BLE001
        print(f"[db-debug] Connection failed: {exc}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())