Your Name
80d73d1b74
- redis/client.py: use BlockingConnectionPool.from_url() (fixes 'url' kwarg error) - database/engine_compat.py: patch aiomysql do_ping to satisfy pool_pre_ping - database/session.py: apply engine_compat patch before create_async_engine - requirements.txt: SQLAlchemy 2.0.49 (fixes aiomysql ping() reconnect signature) - .env.example: document NEXUS_REDIS_URL format - scripts/wsl_ensure_venv.sh: create project .venv (PEP 668 safe) - scripts/wsl_start_dev.sh: use .venv python, validate Redis before start - scripts/wsl_integration_smoke.sh: code-only verification mode - scripts/wsl_check_mysql.py / bootstrap_database.py: MySQL setup helpers - scripts/sync_frontend_vendor.py: vendor asset sync utility Co-authored-by: Cursor <cursoragent@cursor.com>
67 lines
2.1 KiB
Python
67 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Quick MySQL connectivity check (run from WSL: python3 scripts/wsl_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))
|
|
|
|
# Load .env manually for standalone script
|
|
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()))
|