DBUtils

3.1.2 · active · verified Fri Apr 10

DBUtils is a suite of Python modules providing robust, persistent, and pooled connections to a database, designed for multi-threaded environments. It supports DB-API 2 compliant database interfaces and the classic PyGreSQL interface. The current version, 3.1.2, is actively maintained and supports Python versions 3.7 to 3.14.

Warnings

breaking Version 2.0 introduced significant breaking changes. Users upgrading from versions 1.x should carefully review the changelog to understand necessary modifications to their code.
Fix: Consult the official DBUtils changelog and documentation for migration guidance when upgrading from pre-2.0 versions.
gotcha The `dbutils.simple_pooled_db.SimplePooledDB` class is provided as a basic reference implementation and is explicitly NOT recommended for production use due to lacking sophisticated features like failover.
Fix: For production environments, always use `dbutils.pooled_db.PooledDB` or `dbutils.persistent_db.PersistentDB` for robust connection management.
gotcha DBUtils itself does not include a database driver. You must separately install a DB-API 2 compliant driver for your specific database (e.g., `psycopg2` for PostgreSQL, `mysql-connector-python` for MySQL, `cx_Oracle` for Oracle) and pass its `connect` function to `PooledDB` or `PersistentDB`.
Fix: Ensure the appropriate database driver is installed via pip (e.g., `pip install psycopg2-binary`) and imported into your application.

Install

pip install DBUtils Install DBUtils

Imports

PooledDB
```
from dbutils.pooled_db import PooledDB
```

PersistentDB

from dbutils.persistent_db import PersistentDB

SimplePooledDB
```
from dbutils.simple_pooled_db import SimplePooledDB
```
Not recommended for production use; intended as a basic reference implementation.

Quickstart

This quickstart demonstrates how to set up and use `PooledDB` with an in-memory SQLite database. Replace `sqlite3.connect` with your specific DB-API 2 compliant database connector (e.g., `psycopg2.connect`) and provide appropriate connection arguments for your database.

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.")

view raw JSON →