Spaces:
Paused
Paused
File size: 3,033 Bytes
5111c4b |
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 77 78 79 80 81 82 83 84 85 86 87 |
import asyncio
import asyncpg
import ssl
import platform
import sys
import logging
from typing import Optional
# Configure logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)
async def get_ssl_context() -> ssl.SSLContext:
"""Create an SSL context for database connection."""
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
return ssl_context
async def test_connection() -> None:
"""Test the database connection with detailed logging."""
logger.info("Starting database connection test...")
if platform.system() == 'Windows':
logger.info("Configuring Windows-specific event loop policy...")
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
try:
ssl_context = await get_ssl_context()
logger.info("Attempting to establish database connection...")
conn: Optional[asyncpg.Connection] = None
try:
conn = await asyncpg.connect(
user='postgres',
password='Lovyelias5584.',
database='postgres',
host='juycnkjuzylnbruwaqmp.supabase.co',
port=5432,
ssl=ssl_context,
command_timeout=30.0,
server_settings={'application_name': 'pos_backend_test'}
)
logger.info("Connection established successfully")
# Test the connection with a simple query
result = await conn.fetchval('SELECT version()')
logger.info(f"Database version: {result}")
# Test table access
tables = await conn.fetch("""
SELECT table_name
FROM information_schema.tables
WHERE table_schema = 'public'
""")
logger.info("Available tables:")
for table in tables:
logger.info(f"- {table['table_name']}")
except Exception as e:
logger.error(f"Database connection failed: {str(e)}", exc_info=True)
raise
finally:
if conn:
logger.info("Closing database connection...")
await conn.close()
logger.info("Database connection closed")
except Exception as e:
logger.error(f"Test failed: {str(e)}", exc_info=True)
raise
else:
logger.info("Database connection test completed successfully")
if __name__ == "__main__":
try:
asyncio.run(test_connection())
except KeyboardInterrupt:
logger.info("Test interrupted by user")
except Exception as e:
logger.error("Fatal error occurred", exc_info=True)
sys.exit(1) |