107 lines
3.3 KiB
Python
107 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Check Billing Database Connection - Python Version
|
|
|
|
Replacement for check_billing_db.js with same functionality.
|
|
|
|
Usage:
|
|
python check_billing_db.py
|
|
|
|
Environment variables required:
|
|
BILLING_DB_HOST - Database host address
|
|
BILLING_DB_USER - Database username
|
|
BILLING_DB_PASS - Database password
|
|
BILLING_DB_NAME - Database name
|
|
|
|
Example .env file:
|
|
BILLING_DB_HOST=localhost
|
|
BILLING_DB_USER=root
|
|
BILLING_DB_PASS=password
|
|
BILLING_DB_NAME=billing_db
|
|
|
|
Install dependencies:
|
|
pip install -r requirements.txt
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import pymysql
|
|
from pymysql import MySQLError
|
|
from dotenv import load_dotenv
|
|
|
|
def check_billing():
|
|
"""Test connection to BILLING Database"""
|
|
print('Testing connection to BILLING Database...')
|
|
|
|
# Load environment variables
|
|
load_dotenv()
|
|
|
|
# Get environment variables
|
|
host = os.getenv('BILLING_DB_HOST')
|
|
user = os.getenv('BILLING_DB_USER')
|
|
password = os.getenv('BILLING_DB_PASS')
|
|
database = os.getenv('BILLING_DB_NAME')
|
|
|
|
print(f'Host: {host}')
|
|
print(f'User: {user}')
|
|
print(f'Database: {database}')
|
|
|
|
if not all([host, user, password, database]):
|
|
print('ERROR: Missing required environment variables')
|
|
print('Required: BILLING_DB_HOST, BILLING_DB_USER, BILLING_DB_PASS, BILLING_DB_NAME')
|
|
sys.exit(1)
|
|
|
|
try:
|
|
# Validate connection parameters
|
|
if not all([host, user, password, database]):
|
|
print('ERROR: Missing database connection parameters')
|
|
return False
|
|
|
|
# Type assurance after validation
|
|
assert host is not None and user is not None and password is not None and database is not None
|
|
|
|
# Establish connection with timeout
|
|
connection = pymysql.connect(
|
|
host=host,
|
|
user=user,
|
|
password=password,
|
|
database=database,
|
|
connect_timeout=10 # 10 seconds timeout
|
|
)
|
|
|
|
print('Connection successful!')
|
|
|
|
# Test SELECT query
|
|
with connection.cursor(pymysql.cursors.DictCursor) as cursor:
|
|
cursor.execute('SELECT COUNT(*) as count FROM customer')
|
|
rows = cursor.fetchall()
|
|
|
|
if rows:
|
|
count = rows[0]['count']
|
|
print(f'Test SELECT successful. Customer Count: {count}')
|
|
else:
|
|
print('Test SELECT returned no results')
|
|
|
|
connection.close()
|
|
|
|
print('BILLING DB TEST PASSED')
|
|
return True
|
|
|
|
except MySQLError as e:
|
|
print(f'BILLING DB Test Failed: {e}')
|
|
|
|
# Provide helpful hints based on error
|
|
if 'timed out' in str(e).lower() or 'timeout' in str(e).lower():
|
|
print('Hint: Check firewall or IP whitelist on the remote server.')
|
|
elif 'access denied' in str(e).lower():
|
|
print('Hint: Check username/password or database permissions.')
|
|
elif 'unknown database' in str(e).lower():
|
|
print('Hint: Check database name or create the database first.')
|
|
elif 'cannot connect' in str(e).lower():
|
|
print('Hint: Check host/port or network connectivity.')
|
|
|
|
return False
|
|
|
|
if __name__ == '__main__':
|
|
success = check_billing()
|
|
sys.exit(0 if success else 1) |