Nexus/scripts/check_mysql.py

#!/usr/bin/env python3
"""Quick MySQL connectivity check (run: python3 scripts/check_mysql.py)."""
import asyncio
import os
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

env_path = ROOT / ".env"
if env_path.exists():
    for line in env_path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if line and not line.startswith("#") and "=" in line:
            k, v = line.split("=", 1)
            os.environ.setdefault(k.strip(), v.strip())

HOST = os.environ.get("MYSQL_HOST", "127.0.0.1")
PORT = int(os.environ.get("MYSQL_PORT", "3306"))
USER = os.environ.get("MYSQL_USER", "root")
PASSWORD = os.environ.get("MYSQL_PASSWORD", "")
DB = os.environ.get("MYSQL_DATABASE", "nexus")


async def main() -> int:
    import aiomysql

    print(f"Connecting {USER}@{HOST}:{PORT} ...")
    try:
        conn = await aiomysql.connect(
            host=HOST, port=PORT, user=USER, password=PASSWORD, connect_timeout=15
        )
        async with conn.cursor() as cur:
            await cur.execute("SELECT VERSION()")
            print("VERSION:", (await cur.fetchone())[0])
            await cur.execute("SHOW DATABASES")
            dbs = [r[0] for r in await cur.fetchall()]
            print("DATABASES:", dbs)
        conn.close()
    except Exception as e:
        print("FAIL (server):", type(e).__name__, e)
        return 1

    if DB not in dbs:
        print(f"DATABASE '{DB}' missing — run: python3 scripts/bootstrap_database.py")
        return 2

    try:
        conn = await aiomysql.connect(
            host=HOST, port=PORT, user=USER, password=PASSWORD, db=DB, connect_timeout=15
        )
        async with conn.cursor() as cur:
            await cur.execute("SHOW TABLES")
            tables = sorted(r[0] for r in await cur.fetchall())
            print(f"TABLES in {DB} ({len(tables)}):", tables)
        conn.close()
        return 0 if tables else 3
    except Exception as e:
        print(f"FAIL ({DB}):", type(e).__name__, e)
        return 1


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))