Codex Release Bot
66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
#!/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()))
|