DB2 Connector

Connect to and interact with IBM DB2 databases using Python.

Basic Connection

import ibm_db
import ibm_db_dbi

# Connect using connection string
conn_str = (
    "DATABASE=SAMPLE;"
    "HOSTNAME=localhost;"
    "PORT=50000;"
    "PROTOCOL=TCPIP;"
    "UID=db2user;"
    "PWD=password;"
)

# Low-level connection
conn = ibm_db.connect(conn_str, "", "")

# DB-API connection (recommended)
conn = ibm_db_dbi.connect(conn_str)
cursor = conn.cursor()

Executing Queries

# Simple query
cursor.execute("SELECT * FROM employees WHERE dept = ?", ('IT',))
rows = cursor.fetchall()

for row in rows:
    print(row)

# With parameters
sql = "SELECT emp_id, name, salary FROM employees WHERE salary > ?"
cursor.execute(sql, (50000,))
results = cursor.fetchall()

Insert and Update

# Insert data
insert_sql = "INSERT INTO inventory (item_id, quantity, price) VALUES (?, ?, ?)"
cursor.execute(insert_sql, ('ITEM001', 100, 29.99))
conn.commit()

# Batch insert
items = [
    ('ITEM002', 50, 19.99),
    ('ITEM003', 75, 39.99),
    ('ITEM004', 120, 15.99)
]
cursor.executemany(insert_sql, items)
conn.commit()

# Update
update_sql = "UPDATE inventory SET quantity = quantity + ? WHERE item_id = ?"
cursor.execute(update_sql, (25, 'ITEM001'))
conn.commit()

Transactions

try:
    cursor.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", (100, 1))
    cursor.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", (100, 2))
    conn.commit()
except Exception as e:
    conn.rollback()
    print(f"Transaction failed: {e}")

Stored Procedures

# Call stored procedure
call_sql = "CALL calculate_totals(?, ?)"
cursor.execute(call_sql, ('2024', 'Q1'))

# Get results
results = cursor.fetchall()

Connection Pooling

class DB2ConnectionPool:
    def __init__(self, conn_str, pool_size=5):
        self.conn_str = conn_str
        self.pool = []
        for _ in range(pool_size):
            conn = ibm_db_dbi.connect(conn_str)
            self.pool.append(conn)

    def get_connection(self):
        if self.pool:
            return self.pool.pop()
        return ibm_db_dbi.connect(self.conn_str)

    def return_connection(self, conn):
        self.pool.append(conn)

# Usage
pool = DB2ConnectionPool(conn_str)
conn = pool.get_connection()
cursor = conn.cursor()
# ... use cursor ...
pool.return_connection(conn)

Query Results as Dictionaries

def query_as_dict(cursor, sql, params=None):
    """Execute query and return results as dictionaries."""
    cursor.execute(sql, params or ())
    columns = [desc[0] for desc in cursor.description]
    results = []
    for row in cursor.fetchall():
        results.append(dict(zip(columns, row)))
    return results

# Usage
sql = "SELECT item_id, quantity, price FROM inventory"
items = query_as_dict(cursor, sql)
for item in items:
    print(f"Item {item['item_id']}: {item['quantity']} @ ${item['price']}")

Error Handling

import ibm_db_dbi

try:
    conn = ibm_db_dbi.connect(conn_str)
    cursor = conn.cursor()
    cursor.execute(sql)
except ibm_db_dbi.DatabaseError as e:
    print(f"Database error: {e}")
except ibm_db_dbi.ProgrammingError as e:
    print(f"Programming error: {e}")
finally:
    if cursor:
        cursor.close()
    if conn:
        conn.close()

Context Manager

from contextlib import contextmanager

@contextmanager
def db2_connection(conn_str):
    """Context manager for DB2 connections."""
    conn = ibm_db_dbi.connect(conn_str)
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

# Usage
with db2_connection(conn_str) as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM inventory")
    results = cursor.fetchall()