helpers/postgres.py

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