import os import psycopg2 from psycopg2 import pool, OperationalError from dotenv import load_dotenv from typing import Optional, Union, List, Dict, Tuple import psycopg2.extensions # Load environment variables load_dotenv() def log_method(func): def wrapper(*args, **kwargs): print(f"Calling method {func.__name__}") return func(*args, **kwargs) return wrapper class DatabaseConnection: _instance = None # For Singleton pattern (optional) def __init__(self, db_user, db_password, db_host, db_port, db_name): self.db_user = db_user self.db_password = db_password self.db_host = db_host self.db_port = db_port self.db_name = db_name try: # Create a connection pool (min and max connection count) self.connection_pool = psycopg2.pool.SimpleConnectionPool( 1, 10, # Min and max number of connections user=db_user, password=db_password, host=db_host, port=db_port, database=db_name ) if self.connection_pool: print("Connection pool created successfully") except OperationalError as e: print(f"Error while connecting to PostgreSQL: {e}") def execute_query(self, query: str, params: Optional[Union[List, Dict]] = None, connection: Optional[psycopg2.extensions.connection] = None) -> List[Tuple]: cursor = self.connection_pool.getconn().cursor() cursor.execute(query, params) return cursor