scripts/check_live_db.py

"""Quick live DB table + row count check"""
import asyncio, os, sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from dotenv import load_dotenv
load_dotenv(Path(__file__).parent.parent / ".env")

async def main():
    from sqlalchemy.ext.asyncio import create_async_engine
    from sqlalchemy import text
    db_url = os.getenv("DATABASE_URL")
    print("Connecting to DB...")
    engine = create_async_engine(db_url, echo=False)
    async with engine.connect() as conn:
        tables_q = await conn.execute(text(
            "SELECT table_name FROM information_schema.tables WHERE table_schema='public' ORDER BY table_name"
        ))
        tables = [row[0] for row in tables_q]
        print(f"\nFound {len(tables)} tables:\n")
        for tbl in tables:
            cnt_q = await conn.execute(text(f'SELECT COUNT(*) FROM "{tbl}"'))
            cnt = cnt_q.scalar()
            print(f"  {tbl:<35} {cnt} rows")
        
        print("\n\nColumn details:\n")
        for tbl in tables:
            cols_q = await conn.execute(text(
                f"SELECT column_name, data_type FROM information_schema.columns "
                f"WHERE table_schema='public' AND table_name='{tbl}' ORDER BY ordinal_position"
            ))
            cols = cols_q.fetchall()
            print(f"\n  TABLE: {tbl}")
            for c in cols:
                print(f"    - {c[0]:<35} {c[1]}")

    await engine.dispose()
    print("\nDone.")

asyncio.run(main())