psycopg2-pool

import os from psycopg2_pool import ConnectionPool import psycopg2 # Configure your database connection details # Use environment variables for sensitive info in production DB_HOST = os.environ.get('DB_HOST', 'localhost') DB_NAME = os.environ.get('DB_NAME', 'testdb') DB_USER = os.environ.get('DB_USER', 'user') DB_PASSWORD = os.environ.get('DB_PASSWORD', 'password') # Initialize a connection pool with min/max connections # Pass psycopg2.connect keyword arguments directly. pool = None # Initialize pool to None for safe cleanup try: pool = ConnectionPool( minconn=1, maxconn=10, host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASSWORD ) # Example 1: Using 'with' statement for automatic connection return with pool.getconn() as conn: with conn.cursor() as cur: cur.execute("SELECT 1 + 1 AS result;") result = cur.fetchone()[0] print(f"Query result (with statement): {result}") conn.commit() # Commit if changes were made # Example 2: Explicitly getting and returning a connection conn = pool.getconn() try: with conn.cursor() as cur: cur.execute("CREATE TABLE IF NOT EXISTS test_table (id SERIAL PRIMARY KEY, name VARCHAR(50));") conn.commit() print("Table 'test_table' ensured (explicit putconn).") finally: pool.putconn(conn) # Crucial: always return the connection! except psycopg2.Error as e: print(f"Database error: {e}") except Exception as e: print(f"An unexpected error occurred: {e}") finally: # Close the pool when done (e.g., application shutdown) if pool: pool.close() print("Connection pool closed.")