tessl/pypi-cx-oracle

Python interface to Oracle Database implementing DB API 2.0 with Oracle-specific extensions

—

Pending

Overview

Eval results

Files

SQL Execution and Cursors

Name: tessl/pypi-cx-oracle
Author: tessl

SQL statement execution, parameter binding, result fetching, and cursor management with support for all Oracle SQL features and PL/SQL operations.

Capabilities

Cursor Creation and Management

Create and manage cursors for SQL execution and result processing.

class Cursor:
    def __init__(self, connection: Connection):
        """Create cursor from connection (usually via connection.cursor())"""
        
    def close(self) -> None:
        """Close cursor and free resources"""
        
    def prepare(self, sql: str) -> None:
        """Prepare SQL statement for execution"""
        
    def parse(self, sql: str) -> None:
        """Parse SQL statement without execution"""
        
    def __enter__(self) -> Cursor:
        """Context manager entry"""
        
    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        """Context manager exit with automatic cleanup"""

SQL Statement Execution

Execute SQL statements with parameter binding support.

class Cursor:
    def execute(self, sql: str, parameters=None) -> Cursor:
        """
        Execute SQL statement with optional parameters.
        
        Parameters:
        - sql (str): SQL statement to execute
        - parameters: Parameters for SQL statement (dict, list, or tuple)
        
        Returns:
        Self (for method chaining)
        """
        
    def executemany(self, sql: str, seq_of_parameters, **kwargs) -> None:
        """
        Execute SQL statement multiple times with different parameter sets.
        
        Parameters:
        - sql (str): SQL statement to execute
        - seq_of_parameters: Sequence of parameter sets
        - kwargs: Additional execution options
        """
        
    def executemanyprepared(self, numRows: int, **kwargs) -> None:
        """
        Execute many with prepared statement using bind arrays.
        
        Parameters:
        - numRows (int): Number of rows to execute
        - kwargs: Additional execution options
        """

Usage examples:

cursor = connection.cursor()

# Execute with named parameters
cursor.execute("SELECT * FROM employees WHERE department_id = :dept_id", 
               {"dept_id": 10})

# Execute with positional parameters  
cursor.execute("INSERT INTO employees VALUES (:1, :2, :3)", 
               (1001, "John", "Doe"))

# Execute many with parameter sequences
data = [(1002, "Jane", "Smith"), (1003, "Bob", "Jones")]
cursor.executemany("INSERT INTO employees VALUES (:1, :2, :3)", data)

# Execute script
cursor.executescript("""
    CREATE TABLE temp_table (id NUMBER, name VARCHAR2(50));
    INSERT INTO temp_table VALUES (1, 'Test');
    COMMIT;
""")

Result Fetching

Retrieve query results using various fetch methods.

class Cursor:
    def fetchone(self) -> tuple:
        """
        Fetch next row from query results.
        
        Returns:
        Tuple representing row, or None if no more rows
        """
        
    def fetchmany(self, numRows=None) -> list:
        """
        Fetch multiple rows from query results.
        
        Parameters:
        - numRows (int): Number of rows to fetch (default: arraysize)
        
        Returns:
        List of tuples representing rows
        """
        
    def fetchall(self) -> list:
        """
        Fetch all remaining rows from query results.
        
        Returns:
        List of tuples representing all remaining rows
        """
        
    def fetchraw(self, numRows=None) -> list:
        """
        Fetch raw data bypassing type conversion.
        
        Parameters:
        - numRows (int): Number of rows to fetch
        
        Returns:
        List of raw data tuples
        """

Usage examples:

cursor.execute("SELECT employee_id, first_name, last_name FROM employees")

# Fetch one row at a time
row = cursor.fetchone()
while row:
    print(f"ID: {row[0]}, Name: {row[1]} {row[2]}")
    row = cursor.fetchone()

# Fetch multiple rows
cursor.execute("SELECT * FROM departments")
rows = cursor.fetchmany(5)
for row in rows:
    print(row)

# Fetch all rows
cursor.execute("SELECT * FROM employees")
all_rows = cursor.fetchall()
print(f"Total employees: {len(all_rows)}")

# Iterator interface
cursor.execute("SELECT * FROM employees")
for row in cursor:
    print(row)

PL/SQL Execution

Execute stored procedures and functions.

class Cursor:
    def callfunc(self, name: str, returnType, parameters=None):
        """
        Call stored function and return result.
        
        Parameters:
        - name (str): Function name
        - returnType: Expected return type (cx_Oracle type or Python type)
        - parameters: Function parameters (list or tuple)
        
        Returns:
        Function return value
        """
        
    def callproc(self, name: str, parameters=None) -> list:
        """
        Call stored procedure.
        
        Parameters:
        - name (str): Procedure name  
        - parameters: Procedure parameters (list or tuple)
        
        Returns:
        List of parameter values (including OUT parameters)
        """

Usage examples:

# Call function
result = cursor.callfunc("get_employee_salary", cx_Oracle.NUMBER, [1001])
print(f"Salary: {result}")

# Call procedure with OUT parameters
params = [1001, None, None]  # emp_id, OUT first_name, OUT last_name
cursor.callproc("get_employee_info", params)
print(f"Employee: {params[1]} {params[2]}")

# Call procedure with IN/OUT parameters
balance = [1000]  # Initial balance
cursor.callproc("update_account_balance", [12345, 500, balance])
print(f"New balance: {balance[0]}")

Variable Binding and Type Handling

Manage bind variables and data type conversion.

class Cursor:
    def var(self, typ, size=None, arraysize=None, inconverter=None, 
           outconverter=None) -> Var:
        """
        Create bind variable of specified type.
        
        Parameters:
        - typ: Variable type (cx_Oracle type)
        - size (int): Variable size
        - arraysize (int): Array size for array variables
        - inconverter: Input conversion function  
        - outconverter: Output conversion function
        
        Returns:
        Variable object
        """
        
    def arrayvar(self, typ, value, size=None, **kwargs) -> Var:
        """
        Create array bind variable.
        
        Parameters:
        - typ: Variable type (cx_Oracle type)
        - value: Initial array value
        - size (int): Variable size
        - kwargs: Additional variable options
        
        Returns:
        Array variable object
        """
        
    def setinputsizes(self, **kwargs) -> None:
        """
        Set input variable sizes for subsequent execute operations.
        
        Parameters:
        - kwargs: Named parameter sizes (name=type or name=(type, size))
        """
        
    def setoutputsize(self, size: int, column=None) -> None:
        """
        Set output size for large columns.
        
        Parameters:
        - size (int): Output buffer size
        - column: Column index or name (None for all columns)
        """
        
    def bindnames(self) -> list:
        """
        Get names of bind variables in prepared statement.
        
        Returns:
        List of bind variable names
        """
        
    def getbatcherrors(self) -> list:
        """
        Get batch execution errors from executemany operation.
        
        Returns:
        List of error objects for failed rows
        """
        
    def getarraydmlrowcounts(self) -> list:
        """
        Get row counts for each row in array DML operation.
        
        Returns:
        List of row counts for array DML
        """
        
    def getimplicitresults(self) -> list:
        """
        Get implicit result sets from PL/SQL procedure.
        
        Returns:
        List of cursor objects for each result set
        """

Usage examples:

# Create bind variables
var_id = cursor.var(cx_Oracle.NUMBER)
var_name = cursor.var(cx_Oracle.STRING, 100)

# Bind variables to statement
cursor.execute("SELECT first_name INTO :name FROM employees WHERE employee_id = :id",
               {"id": var_id, "name": var_name})

# Set input sizes for better performance
cursor.setinputsizes(name=100, salary=cx_Oracle.NUMBER)
cursor.execute("INSERT INTO employees (name, salary) VALUES (:name, :salary)",
               {"name": "John Doe", "salary": 50000})

# Handle batch errors
try:
    cursor.executemany("INSERT INTO employees VALUES (:1, :2)", data)
except cx_Oracle.Error:
    errors = cursor.getbatcherrors()
    for error in errors:
        print(f"Row {error.offset}: {error.message}")

Cursor Properties

Access cursor metadata and configuration.

class Cursor:
    @property
    def connection(self) -> Connection:
        """Associated connection object"""
        
    @property
    def description(self) -> list:
        """Column descriptions for last query (list of 7-tuples)"""
        
    @property
    def rowcount(self) -> int:
        """Number of rows affected by last operation"""
        
    @property
    def arraysize(self) -> int:
        """Number of rows to fetch at once (default 100)"""
        
    @property
    def bindarraysize(self) -> int:
        """Array size for bind operations"""
        
    @property
    def bindvars(self) -> dict:
        """Dictionary of bind variables"""
        
    @property
    def fetchvars(self) -> list:
        """List of fetch variables"""
        
    @property
    def inputtypehandler(self):
        """Input type handler function"""
        
    @property
    def outputtypehandler(self):
        """Output type handler function"""
        
    @property
    def rowfactory(self):
        """Row factory function for result formatting"""
        
    @property
    def scrollable(self) -> bool:
        """Whether cursor is scrollable"""
        
    @property
    def prefetchrows(self) -> int:
        """Number of rows to prefetch"""
        
    @property
    def statement(self) -> str:
        """Last executed SQL statement"""
        
    @property
    def statementtype(self) -> int:
        """Type of last executed statement"""

Advanced Cursor Features

Advanced cursor operations including scrolling and decimal handling.

class Cursor:
    def scroll(self, value: int, mode='relative') -> None:
        """
        Scroll cursor position (requires scrollable cursor).
        
        Parameters:
        - value (int): Number of rows to scroll
        - mode (str): 'relative' or 'absolute'
        """
        
    def setdecimal(self, precision: int, scale: int) -> None:
        """
        Set decimal precision and scale for NUMBER columns.
        
        Parameters:
        - precision (int): Total number of digits
        - scale (int): Number of digits after decimal point
        """

Usage examples:

# Create scrollable cursor
cursor = connection.cursor(scrollable=True)
cursor.execute("SELECT * FROM employees ORDER BY employee_id")

# Scroll through results
cursor.scroll(10)  # Skip 10 rows
row = cursor.fetchone()

cursor.scroll(-5)  # Go back 5 rows
row = cursor.fetchone()

cursor.scroll(0, 'absolute')  # Go to beginning
row = cursor.fetchone()

# Custom row factory
cursor.rowfactory = lambda *args: dict(zip([d[0] for d in cursor.description], args))
cursor.execute("SELECT employee_id, first_name FROM employees")
for row in cursor:
    print(f"ID: {row['EMPLOYEE_ID']}, Name: {row['FIRST_NAME']}")

# Set decimal handling
cursor.setdecimal(10, 2)  # 10 digits total, 2 after decimal
cursor.execute("SELECT salary FROM employees")

Context Management

Cursors support Python's context manager protocol for automatic cleanup:

with connection.cursor() as cursor:
    cursor.execute("SELECT * FROM employees")
    for row in cursor:
        print(row)
    # Cursor automatically closed when exiting context

Statement Types

Statement type constants for identifying SQL statement types:

# Available via cursor.statementtype after execution
STMT_TYPE_UNKNOWN: int
STMT_TYPE_SELECT: int  
STMT_TYPE_UPDATE: int
STMT_TYPE_DELETE: int
STMT_TYPE_INSERT: int
STMT_TYPE_CREATE: int
STMT_TYPE_DROP: int
STMT_TYPE_ALTER: int
STMT_TYPE_BEGIN: int
STMT_TYPE_DECLARE: int

Performance Optimization

Tips for optimal cursor performance:

# Set appropriate array size for bulk operations
cursor.arraysize = 1000

# Use executemany for bulk inserts/updates
cursor.executemany("INSERT INTO table VALUES (:1, :2)", data)

# Prepare statements for repeated execution
cursor.prepare("SELECT * FROM employees WHERE dept_id = :dept_id")
for dept_id in [10, 20, 30]:
    cursor.execute(None, {"dept_id": dept_id})
    
# Use setinputsizes for better performance with large data
cursor.setinputsizes(data=cx_Oracle.CLOB)
cursor.execute("INSERT INTO documents (id, data) VALUES (:id, :data)", 
               {"id": 1, "data": large_text})

Install with Tessl CLI