oracle__skills/db/appdev/python-oracledb.md

Python + Oracle Database

Overview

python-oracledb is the official Oracle Python driver (successor to cx_Oracle). It supports two modes:

• Thin mode (default): pure Python, no Oracle Client libraries required. Supports most features.
• Thick mode: requires Oracle Client (Instant Client or full client). Required for advanced features like Advanced Queuing, Sharding, and some proxy authentication scenarios.

pip install oracledb

---

Connecting

Basic Connection (Thin Mode)

import oracledb

# Easy Connect string
conn = oracledb.connect(
    user="hr",
    password="password",
    dsn="localhost:1521/freepdb1"
)

# TNS alias (requires tnsnames.ora in TNS_ADMIN path)
conn = oracledb.connect(
    user="hr",
    password="password",
    dsn="mydb_high"
)

# Close explicitly or use context manager
with oracledb.connect(user="hr", password="password", dsn="localhost:1521/freepdb1") as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT sysdate FROM dual")
        print(cur.fetchone())

Wallet / mTLS (Autonomous Database)

import oracledb

conn = oracledb.connect(
    user="admin",
    password="password",
    dsn="myatp_high",           # TNS alias from tnsnames.ora in wallet
    config_dir="/path/to/wallet",
    wallet_location="/path/to/wallet",
    wallet_password="wallet_password"  # if the wallet is password-protected
)

Enabling Thick Mode

import oracledb

# Call before any connection — sets the mode for the entire process
oracledb.init_oracle_client(lib_dir="/opt/oracle/instantclient_21_9")

conn = oracledb.connect(user="hr", password="password", dsn="localhost:1521/freepdb1")

---

Executing SQL

Bind Variables

Always use bind variables — never format user input into SQL strings.

with conn.cursor() as cur:
    # Named binds (recommended)
    cur.execute(
        "SELECT last_name, salary FROM employees WHERE department_id = :dept_id AND salary > :min_sal",
        dept_id=60,
        min_sal=5000
    )

    # Or pass as dict
    cur.execute(
        "SELECT last_name FROM employees WHERE employee_id = :id",
        {"id": 100}
    )

    rows = cur.fetchall()
    for row in rows:
        print(row)

DML with Binds

with conn.cursor() as cur:
    cur.execute(
        "UPDATE employees SET salary = :sal WHERE employee_id = :id",
        sal=9000,
        id=100
    )
    conn.commit()

Batch Execution (executemany)

data = [
    {"id": 201, "name": "Alice", "dept": 10},
    {"id": 202, "name": "Bob",   "dept": 20},
    {"id": 203, "name": "Carol", "dept": 10},
]

with conn.cursor() as cur:
    cur.executemany(
        "INSERT INTO employees (employee_id, last_name, department_id) VALUES (:id, :name, :dept)",
        data
    )
    conn.commit()

---

Fetching Results

with conn.cursor() as cur:
    cur.execute("SELECT employee_id, last_name, salary FROM employees WHERE rownum <= 100")

    # fetchone — single row
    row = cur.fetchone()

    # fetchmany — batch
    rows = cur.fetchmany(numRows=25)

    # fetchall — all remaining (careful with large result sets)
    rows = cur.fetchall()

    # Iterate directly (memory-efficient for large sets)
    cur.execute("SELECT * FROM employees")
    for row in cur:
        print(row)

Column Names

with conn.cursor() as cur:
    cur.execute("SELECT employee_id, last_name, salary FROM employees WHERE rownum <= 5")
    columns = [col[0] for col in cur.description]
    rows = cur.fetchall()
    for row in rows:
        print(dict(zip(columns, row)))

Fetch as Dictionaries (rowfactory)

def make_dict_factory(cursor):
    cols = [col[0].lower() for col in cursor.description]
    def create_row(*args):
        return dict(zip(cols, args))
    return create_row

with conn.cursor() as cur:
    cur.execute("SELECT employee_id, last_name FROM employees WHERE rownum <= 5")
    cur.rowfactory = make_dict_factory(cur)
    for row in cur:
        print(row)  # {'employee_id': 100, 'last_name': 'King'}

---

Connection Pooling

Use a connection pool for web applications and multi-threaded code. Never share a single connection across threads.

import oracledb

# Create pool at application startup
pool = oracledb.create_pool(
    user="hr",
    password="password",
    dsn="localhost:1521/freepdb1",
    min=2,       # minimum open connections
    max=10,      # maximum connections
    increment=1  # connections opened when more are needed
)

# Acquire a connection from the pool
with pool.acquire() as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT COUNT(*) FROM employees")
        print(cur.fetchone())

# Pool is returned automatically; close pool at app shutdown
pool.close()

---

PL/SQL Calls

import oracledb

with conn.cursor() as cur:
    # Call a stored procedure
    cur.callproc("hr.update_salary", [100, 9500])
    conn.commit()

    # Call a function
    result = cur.callfunc("hr.get_employee_count", oracledb.DB_TYPE_NUMBER, [10])
    print(result)

    # Anonymous PL/SQL block with OUT parameters
    out_val = cur.var(oracledb.DB_TYPE_VARCHAR)
    cur.execute(
        """
        BEGIN
          :out := 'Hello from PL/SQL';
        END;
        """,
        out=out_val
    )
    print(out_val.getvalue())

REF CURSOR

with conn.cursor() as cur:
    ref_cursor = cur.var(oracledb.DB_TYPE_CURSOR)
    cur.execute(
        """
        BEGIN
          OPEN :rc FOR SELECT employee_id, last_name FROM employees WHERE department_id = :dept;
        END;
        """,
        rc=ref_cursor,
        dept=60
    )
    for row in ref_cursor.getvalue():
        print(row)

---

LOB Handling

import oracledb

# Read a CLOB
with conn.cursor() as cur:
    cur.execute("SELECT resume FROM employee_docs WHERE employee_id = 100")
    row = cur.fetchone()
    if row:
        clob = row[0]
        text = clob.read()      # reads entire CLOB as string
        print(text[:200])

# Write a CLOB
with conn.cursor() as cur:
    large_text = "..." * 10000
    cur.execute(
        "UPDATE employee_docs SET resume = :clob WHERE employee_id = :id",
        clob=large_text,
        id=100
    )
    conn.commit()

# Read a BLOB
with conn.cursor() as cur:
    cur.execute("SELECT photo FROM employee_photos WHERE employee_id = 100")
    row = cur.fetchone()
    if row:
        blob_data = row[0].read()
        with open("photo.jpg", "wb") as f:
            f.write(blob_data)

---

Async Support (python-oracledb 2.x)

import asyncio
import oracledb

async def main():
    conn = await oracledb.connect_async(
        user="hr",
        password="password",
        dsn="localhost:1521/freepdb1"
    )
    async with conn.cursor() as cur:
        await cur.execute("SELECT COUNT(*) FROM employees")
        row = await cur.fetchone()
        print(row)
    await conn.close()

asyncio.run(main())

# Async pool
async def pooled():
    pool = oracledb.create_pool_async(
        user="hr", password="password", dsn="localhost:1521/freepdb1",
        min=2, max=10, increment=1
    )
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT 1 FROM dual")
            print(await cur.fetchone())
    await pool.close()

asyncio.run(pooled())

---

Best Practices

• Always use bind variables — never concatenate user input into SQL.
• Use a connection pool in web/multi-threaded apps; do not share connections across threads.
• Use context managers (with) for connections and cursors to ensure cleanup.
• Use executemany for bulk inserts/updates instead of looping execute.
• Set arraysize on the cursor for large fetches: cur.arraysize = 1000 reduces round-trips.
• Commit explicitly — python-oracledb does not auto-commit.
• Prefer thin mode unless you specifically need thick-mode-only features.

---

Common Mistakes

Mistake	Problem	Fix
f"SELECT ... WHERE id = {user_input}"	SQL injection	Use bind variables
Sharing one connection across threads	Corruption, errors	Use a pool; one connection per thread
Not calling conn.commit()	DML silently rolled back on disconnect	Always commit or use autocommit=True for simple scripts
fetchall() on a million-row query	OOM	Use fetchmany() or iterate the cursor
Using cx_Oracle in new projects	Deprecated	Migrate to python-oracledb
Not setting arraysize for large fetches	Excessive round-trips	cur.arraysize = 1000 before fetch

---

Oracle Version Notes (19c vs 26ai)

• Thin mode is supported for Oracle Database 12.1 and later.
• python-oracledb 2.x adds async support and JSON improvements for 21c+.
• Oracle 23ai JSON Relational Duality Views are accessible via normal SELECT queries.

Sources

• python-oracledb Documentation
• python-oracledb GitHub
• Oracle Python Developer Center
• Migrating from cx_Oracle to python-oracledb