tessl/pypi-oracledb
Python interface to Oracle Database with thin and thick connectivity modes
—
Pending
sql-execution.mddocs/
SQL Execution
SQL and PL/SQL execution through cursors with support for bind variables, batch operations, stored procedures, and functions. Includes comprehensive result fetching and data type handling with both synchronous and asynchronous operation modes.
Capabilities
Cursor Class
Execute SQL statements and fetch results with full control over execution parameters and result handling.
class Cursor:
"""Cursor for executing SQL statements and fetching results."""
# Properties
arraysize: int
description: list
rowcount: int
statement: str
bindnames: list
fetchvars: list
def execute(self, statement, parameters=None) -> None:
"""
Execute a SQL statement with optional bind parameters.
Parameters:
- statement (str): SQL statement to execute
- parameters (dict|list|tuple): Bind parameters
"""
def executemany(self, statement, parameters) -> None:
"""
Execute a SQL statement with multiple parameter sets.
Parameters:
- statement (str): SQL statement to execute
- parameters (list): List of parameter sets
"""
def fetchone(self) -> tuple:
"""
Fetch the next row from the result set.
Returns:
tuple: Single row or None if no more rows
"""
def fetchmany(self, size=None) -> list:
"""
Fetch multiple rows from the result set.
Parameters:
- size (int): Number of rows to fetch (default: arraysize)
Returns:
list: List of tuples representing rows
"""
def fetchall(self) -> list:
"""
Fetch all remaining rows from the result set.
Returns:
list: List of tuples representing all rows
"""
def callfunc(self, name, return_type, parameters=None):
"""
Call a stored function and return its result.
Parameters:
- name (str): Function name
- return_type: Expected return type
- parameters (list): Function parameters
Returns:
Function return value
"""
def callproc(self, name, parameters=None) -> list:
"""
Call a stored procedure.
Parameters:
- name (str): Procedure name
- parameters (list): Procedure parameters
Returns:
list: Modified parameter values
"""
def var(self, data_type, size=None, arraysize=None, inconverter=None, outconverter=None, encoding=None, nencoding=None, bypass_decode=False) -> Var:
"""
Create a bind variable.
Parameters:
- data_type: Variable data type
- size (int): Maximum size
- arraysize (int): Array size for array variables
- inconverter: Input converter function
- outconverter: Output converter function
Returns:
Var: Variable object
"""
def setinputsizes(self, **parameters) -> None:
"""
Set input parameter sizes for better performance.
Parameters:
- **parameters: Parameter names and their types/sizes
"""
def close(self) -> None:
"""Close the cursor and free resources."""
def parse(self, statement) -> None:
"""
Parse a SQL statement without executing it.
Parameters:
- statement (str): SQL statement to parse
"""
def scroll(self, value, mode="relative") -> None:
"""
Scroll the cursor position (scrollable cursors only).
Parameters:
- value (int): Number of rows to scroll
- mode (str): "relative" or "absolute"
"""
def __iter__(self):
"""Iterator interface for fetching rows."""
def __next__(self):
"""Get next row in iteration."""Async Cursor Class
Asynchronous version of Cursor with async/await support for all operations.
class AsyncCursor:
"""Asynchronous cursor for executing SQL statements."""
# Properties (same as Cursor)
arraysize: int
description: list
rowcount: int
statement: str
bindnames: list
fetchvars: list
async def execute(self, statement, parameters=None) -> None:
"""
Execute a SQL statement with optional bind parameters.
Parameters:
- statement (str): SQL statement to execute
- parameters (dict|list|tuple): Bind parameters
"""
async def executemany(self, statement, parameters) -> None:
"""
Execute a SQL statement with multiple parameter sets.
Parameters:
- statement (str): SQL statement to execute
- parameters (list): List of parameter sets
"""
async def fetchone(self) -> tuple:
"""
Fetch the next row from the result set.
Returns:
tuple: Single row or None if no more rows
"""
async def fetchmany(self, size=None) -> list:
"""
Fetch multiple rows from the result set.
Parameters:
- size (int): Number of rows to fetch (default: arraysize)
Returns:
list: List of tuples representing rows
"""
async def fetchall(self) -> list:
"""
Fetch all remaining rows from the result set.
Returns:
list: List of tuples representing all rows
"""
async def callfunc(self, name, return_type, parameters=None):
"""
Call a stored function and return its result.
Parameters:
- name (str): Function name
- return_type: Expected return type
- parameters (list): Function parameters
Returns:
Function return value
"""
async def callproc(self, name, parameters=None) -> list:
"""
Call a stored procedure.
Parameters:
- name (str): Procedure name
- parameters (list): Procedure parameters
Returns:
list: Modified parameter values
"""
async def close(self) -> None:
"""Close the cursor and free resources."""Variable Management
Create and manage bind variables for optimal performance and data type handling.
class Var:
"""Bind variable for SQL parameters."""
def getvalue(self, pos=0):
"""
Get the variable value.
Parameters:
- pos (int): Position for array variables
Returns:
Variable value
"""
def setvalue(self, pos, value) -> None:
"""
Set the variable value.
Parameters:
- pos (int): Position for array variables
- value: Value to set
"""Result Metadata
Access detailed information about query result columns.
class FetchInfo:
"""Column metadata for fetch operations."""
name: str
type_code: int
type: type
display_size: int
internal_size: int
precision: int
scale: int
null_ok: bool
annotations: dict
is_json: bool
vector_dimensions: int
vector_format: intUsage Examples
Basic SQL Execution
import oracledb
# Connect to database
connection = oracledb.connect(user="hr", password="password", dsn="localhost/xepdb1")
with connection.cursor() as cursor:
# Simple query
cursor.execute("SELECT employee_id, first_name, last_name FROM employees WHERE department_id = 10")
# Fetch results
for row in cursor:
print(f"Employee {row[0]}: {row[1]} {row[2]}")
connection.close()Using Bind Variables
import oracledb
connection = oracledb.connect(user="hr", password="password", dsn="localhost/xepdb1")
with connection.cursor() as cursor:
# Named bind variables
cursor.execute("""
SELECT employee_id, first_name, salary
FROM employees
WHERE department_id = :dept_id
AND salary > :min_salary
""", dept_id=10, min_salary=5000)
results = cursor.fetchall()
for row in results:
print(f"Employee {row[0]}: {row[1]}, Salary: ${row[2]}")
# Positional bind variables
cursor.execute("""
SELECT COUNT(*) FROM employees WHERE hire_date >= :1 AND hire_date < :2
""", ['2020-01-01', '2021-01-01'])
count = cursor.fetchone()[0]
print(f"Employees hired in 2020: {count}")
connection.close()Batch Operations
import oracledb
connection = oracledb.connect(user="hr", password="password", dsn="localhost/xepdb1")
with connection.cursor() as cursor:
# Insert multiple rows
data = [
(1001, 'John', 'Doe', 50000),
(1002, 'Jane', 'Smith', 55000),
(1003, 'Bob', 'Johnson', 60000)
]
cursor.executemany("""
INSERT INTO temp_employees (employee_id, first_name, last_name, salary)
VALUES (:1, :2, :3, :4)
""", data)
connection.commit()
print(f"Inserted {cursor.rowcount} rows")
connection.close()Calling Stored Procedures and Functions
import oracledb
connection = oracledb.connect(user="hr", password="password", dsn="localhost/xepdb1")
with connection.cursor() as cursor:
# Call a function
result = cursor.callfunc("calculate_bonus", oracledb.NUMBER, [12000, 0.15])
print(f"Calculated bonus: {result}")
# Call a procedure with IN/OUT parameters
emp_id = cursor.var(oracledb.NUMBER)
emp_id.setvalue(0, 100)
salary = cursor.var(oracledb.NUMBER)
cursor.callproc("get_employee_salary", [emp_id, salary])
print(f"Employee {emp_id.getvalue()} salary: {salary.getvalue()}")
connection.close()Working with Large Result Sets
import oracledb
connection = oracledb.connect(user="hr", password="password", dsn="localhost/xepdb1")
with connection.cursor() as cursor:
# Set fetch array size for better performance
cursor.arraysize = 1000
cursor.execute("SELECT * FROM large_table")
# Process results in chunks
while True:
rows = cursor.fetchmany()
if not rows:
break
# Process chunk of rows
print(f"Processing {len(rows)} rows...")
for row in rows:
# Process each row
pass
connection.close()Async SQL Execution
import asyncio
import oracledb
async def main():
connection = await oracledb.connect_async(user="hr", password="password", dsn="localhost/xepdb1")
async with connection.cursor() as cursor:
# Async query execution
await cursor.execute("SELECT COUNT(*) FROM employees")
result = await cursor.fetchone()
print(f"Total employees: {result[0]}")
# Async batch operations
data = [(f"Name{i}", i * 1000) for i in range(1, 101)]
await cursor.executemany("INSERT INTO temp_table (name, value) VALUES (:1, :2)", data)
await connection.commit()
await connection.close()
asyncio.run(main())Advanced Variable Usage
import oracledb
connection = oracledb.connect(user="hr", password="password", dsn="localhost/xepdb1")
with connection.cursor() as cursor:
# Create variables with specific types and sizes
cursor.setinputsizes(
name=oracledb.STRING,
salary=oracledb.NUMBER,
hire_date=oracledb.DATETIME
)
# Use variables for better performance in loops
name_var = cursor.var(oracledb.STRING)
salary_var = cursor.var(oracledb.NUMBER)
for i in range(100):
name_var.setvalue(0, f"Employee_{i}")
salary_var.setvalue(0, 50000 + i * 100)
cursor.execute("""
INSERT INTO employees_temp (name, salary)
VALUES (:name, :salary)
""", name=name_var, salary=salary_var)
connection.commit()
connection.close()Install with Tessl CLI
npx tessl i tessl/pypi-oracledb