File size: 2,139 Bytes
11026ec |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 |
#!/usr/bin/env python3
"""
Simple script to test database connection (SQLite or PostgreSQL).
"""
import asyncio
import os
from dotenv import load_dotenv
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import text
# Load environment variables
load_dotenv()
DATABASE_URL = os.getenv("DATABASE_URL")
if not DATABASE_URL:
print("❌ DATABASE_URL not found in environment")
exit(1)
# SQLite URLs don't need conversion, PostgreSQL does
if DATABASE_URL.startswith("postgresql://"):
DATABASE_URL = DATABASE_URL.replace("postgresql://", "postgresql+asyncpg://", 1)
print(f"🔗 Connecting to database...")
print(f" URL: {DATABASE_URL}")
async def test_connection():
"""Test database connection."""
try:
# Create engine
engine = create_async_engine(
DATABASE_URL,
echo=True,
pool_pre_ping=True
)
# Test connection
async with engine.begin() as conn:
# Test basic query
result = await conn.execute(text("SELECT 1"))
test = result.scalar()
print(f"\n✅ Connection successful!")
# List existing tables (works for both SQLite and PostgreSQL)
if "sqlite" in DATABASE_URL:
result = await conn.execute(text("""
SELECT name FROM sqlite_master
WHERE type='table'
ORDER BY name
"""))
else:
result = await conn.execute(text("""
SELECT tablename
FROM pg_tables
WHERE schemaname = 'public'
ORDER BY tablename
"""))
tables = result.fetchall()
print(f"\n📋 Existing tables ({len(tables)}):")
for table in tables:
print(f" - {table[0]}")
await engine.dispose()
return True
except Exception as e:
print(f"\n❌ Connection failed: {e}")
return False
if __name__ == "__main__":
success = asyncio.run(test_connection())
exit(0 if success else 1)
|