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>
46 lines
1.3 KiB
Python
46 lines
1.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Show MySQL grants for MYSQL_USER from .env (WSL helper)."""
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def load_env() -> None:
|
|
env_path = ROOT / ".env"
|
|
if not env_path.exists():
|
|
return
|
|
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())
|
|
|
|
|
|
async def main() -> int:
|
|
import aiomysql
|
|
|
|
load_env()
|
|
user = os.environ.get("MYSQL_USER", "Nexus")
|
|
host = os.environ.get("MYSQL_HOST", "127.0.0.1")
|
|
port = int(os.environ.get("MYSQL_PORT", "3306"))
|
|
password = os.environ.get("MYSQL_PASSWORD", "")
|
|
db = os.environ.get("MYSQL_DATABASE", "nexus")
|
|
|
|
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 GRANTS")
|
|
print(f"Grants for {user}@{host}:")
|
|
for row in await cur.fetchall():
|
|
print(" ", row[0])
|
|
conn.close()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(asyncio.run(main()))
|