Spaces:
Sleeping
Sleeping
File size: 1,586 Bytes
6812fa3 |
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 |
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 |