dbutils

import sqlite3 from dbutils.pooled_db import PooledDB # This example uses sqlite3 which is part of Python's standard library. # For other databases, replace sqlite3.connect with your actual DB-API 2 connect function, # e.g., import psycopg2; db_module = psycopg2 # Create a pool of connections # mincached: Minimum number of connections to keep in the pool # maxcached: Maximum number of connections to keep in the pool # maxconnections: Maximum number of connections to create in total # blocking: Whether to block if maxconnections is reached (True) or raise an error (False) # creator: The DB-API 2 module's connect function pool = PooledDB( creator=sqlite3.connect, database=":memory:", # Use an in-memory SQLite database for the example mincached=2, maxcached=5, maxconnections=10, blocking=True ) def get_data_from_db(pool_instance): conn = None cursor = None try: # Get a connection from the pool conn = pool_instance.connection() cursor = conn.cursor() cursor.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)") cursor.execute("INSERT INTO users (name) VALUES ('Alice')") conn.commit() cursor.execute("SELECT * FROM users") rows = cursor.fetchall() print(f"Fetched users: {rows}") return rows except Exception as e: print(f"An error occurred: {e}") if conn: conn.rollback() finally: # Return the connection to the pool if conn: conn.close() if __name__ == "__main__": print("Running quickstart example for DBUtils.PooledDB") get_data_from_db(pool) get_data_from_db(pool) # Get data again, should reuse pooled connections print("Example finished.")