| 1 | #!/usr/bin/env python3 |
| 2 | """ |
| 3 | Database configuration module |
| 4 | """ |
| 5 | |
| 6 | import os |
| 7 | from typing import Dict, Optional |
| 8 | |
| 9 | # Production database settings |
| 10 | DATABASE_CONFIG = { |
| 11 | 'host': 'prod.example.com', |
| 12 | 'port': 5432, |
| 13 | 'database': 'production_db', |
| 14 | 'user': 'prod_user', |
| 15 | 'password': os.environ.get('DB_PASSWORD'), |
| 16 | 'pool_size': 20, |
| 17 | 'timeout': 30 |
| 18 | # Staging database settings |
| 19 | DATABASE_CONFIG = { |
| 20 | 'host': 'staging.example.com', |
| 21 | 'port': 5432, |
| 22 | 'database': 'staging_db', |
| 23 | 'user': 'staging_user', |
| 24 | 'password': os.environ.get('DB_PASSWORD'), |
| 25 | 'pool_size': 10, |
| 26 | 'timeout': 30 |
| 27 | } |
| 28 | |
| 29 | class DatabaseConnection: |
| 30 | def __init__(self, config: Dict): |
| 31 | self.config = config |
| 32 | self.connection = None |
| 33 | |
| 34 | def connect(self): |
| 35 | """Establish database connection with retry logic""" |
| 36 | max_retries = 3 |
| 37 | for attempt in range(max_retries): |
| 38 | try: |
| 39 | self.connection = create_connection(self.config) |
| 40 | print(f"Connected to {self.config['host']} (production)") |
| 41 | return True |
| 42 | except ConnectionError as e: |
| 43 | print(f"Connection attempt {attempt + 1} failed: {e}") |
| 44 | if attempt == max_retries - 1: |
| 45 | raise |
| 46 | return False |
| 47 | """Establish database connection""" |
| 48 | max_retries = 5 |
| 49 | for attempt in range(max_retries): |
| 50 | try: |
| 51 | self.connection = create_connection(self.config) |
| 52 | print(f"Connected to {self.config['host']} (staging)") |
| 53 | return True |
| 54 | except ConnectionError as e: |
| 55 | print(f"Connection attempt {attempt + 1} failed: {e}") |
| 56 | if attempt == max_retries - 1: |
| 57 | raise |
| 58 | return False |
| 59 | |
| 60 | def disconnect(self): |
| 61 | if self.connection: |
| 62 | self.connection.close() |
| 63 | print("Disconnected from production database") |
| 64 | print("Disconnected from staging database") |
| 65 | self.connection = None |
| 66 | |
| 67 | def get_connection() -> Optional[DatabaseConnection]: |
| 68 | """Factory function to create database connection""" |
| 69 | # Use production settings |
| 70 | return DatabaseConnection(DATABASE_CONFIG) |
| 71 | # Use staging settings |
| 72 | return DatabaseConnection(DATABASE_CONFIG) |
| 73 | |
| 74 | if __name__ == '__main__': |
| 75 | conn = get_connection() |
| 76 | if conn.connect(): |
| 77 | print("Database connection successful!") |
| 78 | conn.disconnect() |
| 79 |