or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

advanced-queuing.mdconnection-pooling.mdconnectivity.mddata-types.mddatabase-objects.mdindex.mdlobs.mdpipeline.mdsoda.mdsql-execution.mdsubscriptions.md
tile.json

tessl/pypi-oracledb

Python interface to Oracle Database with thin and thick connectivity modes

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/oracledb@3.3.x

To install, run

npx @tessl/cli install tessl/pypi-oracledb@3.3.0

index.mddocs/

Oracle Database Python Driver (oracledb)

A comprehensive Python interface to Oracle Database that conforms to the Python Database API 2.0 specification with extensive additions. Supports both thin mode (direct connection without client libraries) and thick mode (using Oracle Client libraries for advanced functionality). The library enables Python programs to execute SQL and PL/SQL statements, call NoSQL-style document APIs, work with data frames, receive database notifications and messages, and perform database administration tasks.

Package Information

  • Package Name: oracledb
  • Language: Python
  • Installation: pip install oracledb
  • Documentation: http://python-oracledb.readthedocs.io

Core Imports

import oracledb

For specific functionality:

from oracledb import connect, create_pool, Connection, Cursor

Basic Usage

import oracledb
import getpass

# Basic connection
username = "scott"
dsn = "localhost/orclpdb"
password = getpass.getpass(f"Enter password for {username}@{dsn}: ")

# Connect and execute SQL
with oracledb.connect(user=username, password=password, dsn=dsn) as connection:
    with connection.cursor() as cursor:
        # Execute a query
        cursor.execute("SELECT sysdate FROM dual")
        result = cursor.fetchone()
        print(f"Current date: {result[0]}")
        
        # Execute with parameters
        cursor.execute("SELECT employee_id, first_name FROM employees WHERE department_id = :dept_id", 
                      dept_id=10)
        for row in cursor.fetchall():
            print(f"Employee {row[0]}: {row[1]}")

Architecture

Oracle python-oracledb provides a comprehensive database interface with two connectivity modes:

  • Thin Mode: Direct connection to Oracle Database without requiring client libraries, suitable for most applications
  • Thick Mode: Uses Oracle Client libraries, providing additional functionality and performance optimizations
  • Connection Management: Single connections and connection pools for scalable applications
  • Execution Model: Cursors for SQL execution with support for bind variables, batch operations, and result fetching
  • Data Handling: Support for all Oracle data types including LOBs, objects, collections, and specialized types like JSON and vectors
  • Advanced Features: Pipelining for performance, event subscriptions, two-phase commit, and administrative operations

Capabilities

Database Connectivity

Core connection management including single connections, connection pools, and connection parameter configuration. Supports both synchronous and asynchronous operations with thin and thick connectivity modes.

def connect(user=None, password=None, dsn=None, **kwargs) -> Connection: ...
def connect_async(user=None, password=None, dsn=None, **kwargs) -> AsyncConnection: ...
def create_pool(user=None, password=None, dsn=None, **kwargs) -> ConnectionPool: ...
def create_pool_async(user=None, password=None, dsn=None, **kwargs) -> AsyncConnectionPool: ...

class Connection:
    def cursor(self, scrollable=False) -> Cursor: ...
    def commit(self) -> None: ...
    def rollback(self) -> None: ...
    def close(self) -> None: ...

class AsyncConnection:
    async def cursor(self, scrollable=False) -> AsyncCursor: ...
    async def commit(self) -> None: ...
    async def rollback(self) -> None: ...
    async def close(self) -> None: ...

Database Connectivity

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.

class Cursor:
    def execute(self, statement, parameters=None) -> None: ...
    def executemany(self, statement, parameters) -> None: ...
    def fetchone(self) -> tuple: ...
    def fetchmany(self, size=None) -> list: ...
    def fetchall(self) -> list: ...
    def callfunc(self, name, return_type, parameters=None): ...
    def callproc(self, name, parameters=None) -> list: ...

class AsyncCursor:
    async def execute(self, statement, parameters=None) -> None: ...
    async def executemany(self, statement, parameters) -> None: ...
    async def fetchone(self) -> tuple: ...
    async def fetchmany(self, size=None) -> list: ...
    async def fetchall(self) -> list: ...

SQL Execution

Connection Pooling

Manage pools of database connections for scalable applications. Supports configuration of pool size, connection timeout, and connection acquisition modes.

class ConnectionPool:
    def acquire(self, user=None, password=None, cclass=None, purity=None, tag=None, matchanytag=False, shardingkey=None, supershardingkey=None) -> Connection: ...
    def release(self, connection, tag=None) -> None: ...
    def close(self, force=False) -> None: ...
    def reconfigure(self, min=None, max=None, increment=None, **kwargs) -> None: ...

class AsyncConnectionPool:
    async def acquire(self, user=None, password=None, cclass=None, purity=None, tag=None, matchanytag=False, shardingkey=None, supershardingkey=None) -> AsyncConnection: ...
    async def release(self, connection, tag=None) -> None: ...
    async def close(self, force=False) -> None: ...

Connection Pooling

Large Objects (LOBs)

Handle Binary Large Objects (BLOB), Character Large Objects (CLOB), National Character Large Objects (NCLOB), and Binary Files (BFILE) with streaming read/write operations.

class LOB:
    def read(self, offset=1, amount=None) -> bytes | str: ...
    def write(self, data, offset=1) -> None: ...
    def size(self) -> int: ...
    def trim(self, new_size) -> None: ...
    def getchunksize(self) -> int: ...
    def open(self) -> None: ...
    def close(self) -> None: ...

class AsyncLOB:
    async def read(self, offset=1, amount=None) -> bytes | str: ...
    async def write(self, data, offset=1) -> None: ...
    async def size(self) -> int: ...
    async def trim(self, new_size) -> None: ...

Large Objects

Database Objects

Work with Oracle user-defined types including objects, collections, and nested tables. Provides object type metadata and instance manipulation.

class DbObject:
    def asdict(self) -> dict: ...
    def aslist(self) -> list: ...
    def copy(self) -> DbObject: ...
    def append(self, element) -> None: ...
    def extend(self, sequence) -> None: ...
    def delete(self, index) -> None: ...
    def exists(self, index) -> bool: ...
    def getelement(self, index): ...
    def setelement(self, index, value) -> None: ...
    def size(self) -> int: ...
    def trim(self, size) -> None: ...

class DbObjectType:
    def newobject(self, value=None) -> DbObject: ...

Database Objects

Pipeline Operations

Batch multiple database operations for improved performance using pipelining. Supports execute, fetch, and stored procedure operations in batches.

class Pipeline:
    def add_execute(self, statement, parameters=None) -> None: ...
    def add_executemany(self, statement, parameters) -> None: ...
    def add_fetchall(self) -> None: ...
    def add_fetchone(self) -> None: ...
    def add_fetchmany(self, size=None) -> None: ...
    def add_callfunc(self, name, return_type, parameters=None) -> None: ...
    def add_callproc(self, name, parameters=None) -> None: ...
    def add_commit(self) -> None: ...
    def execute(self) -> list: ...

def create_pipeline() -> Pipeline: ...

Pipeline Operations

Simple Oracle Document Access (SODA)

NoSQL-style API for working with JSON documents in Oracle Database. SODA provides document operations, filtering, indexing, and metadata management without writing SQL.

# Access SODA through connection
def getSodaDatabase(self) -> SodaDatabase: ...

class SodaDatabase:
    def createCollection(self, name: str, metadata: Union[str, dict] = None, mapMode: bool = False) -> SodaCollection: ...
    def createDocument(self, content: Any, key: str = None, mediaType: str = "application/json") -> SodaDocument: ...
    def getCollectionNames(self, startName: str = None, limit: int = 0) -> List[str]: ...
    def openCollection(self, name: str) -> SodaCollection: ...

class SodaCollection:
    def find(self) -> SodaOperation: ...
    def insertOne(self, doc: Any) -> None: ...
    def insertMany(self, docs: list) -> None: ...
    def save(self, doc: Any) -> None: ...
    def drop(self) -> bool: ...
    def createIndex(self, spec: Union[dict, str]) -> None: ...

class SodaOperation:
    def filter(self, value: Union[dict, str]) -> SodaOperation: ...
    def key(self, value: str) -> SodaOperation: ...
    def limit(self, value: int) -> SodaOperation: ...
    def getDocuments(self) -> list: ...
    def getOne(self) -> Union[SodaDocument, None]: ...
    def remove(self) -> int: ...
    def replaceOne(self, doc: Any) -> bool: ...

Simple Oracle Document Access

Advanced Queuing (AQ)

Oracle Advanced Queuing provides message queuing functionality for reliable, persistent, and transactional message passing using the database as a message broker.

# Access queues through connection
def queue(self, name: str, payload_type: DbObjectType = None) -> Queue: ...

class Queue:
    def enqone(self, message: MessageProperties) -> None: ...
    def enqmany(self, messages: list) -> None: ...
    def deqone(self) -> Union[MessageProperties, None]: ...
    def deqmany(self, max_num_messages: int) -> list: ...

class AsyncQueue:
    async def enqone(self, message: MessageProperties) -> None: ...
    async def enqmany(self, messages: list) -> None: ...
    async def deqone(self) -> Union[MessageProperties, None]: ...
    async def deqmany(self, max_num_messages: int) -> list: ...

class MessageProperties:
    payload: Union[bytes, str, dict, list, DbObject]
    priority: int
    correlation: str
    delay: int
    expiration: int

class DeqOptions:
    mode: int  # DEQ_BROWSE, DEQ_LOCKED, DEQ_REMOVE, etc.
    navigation: int
    visibility: int
    wait: int

class EnqOptions:
    visibility: int
    deliverymode: int

Advanced Queuing

Event Subscriptions

Subscribe to database events including object changes, query result changes, and Advanced Queuing (AQ) messages for real-time notifications.

class Message:
    type: int
    dbname: str
    tables: list
    queries: list
    consumer_name: str
    queue_name: str

class MessageTable:
    name: str
    operation: int
    rows: list

class MessageRow:
    operation: int
    rowid: str

Event Subscriptions

Data Types and Constants

Comprehensive support for all Oracle data types, authentication modes, and configuration constants.

# Database Type Constants
DB_TYPE_VARCHAR: type
DB_TYPE_NUMBER: type
DB_TYPE_DATE: type
DB_TYPE_TIMESTAMP: type
DB_TYPE_BLOB: type
DB_TYPE_CLOB: type
DB_TYPE_JSON: type
DB_TYPE_VECTOR: type

# Authentication Modes
AUTH_MODE_DEFAULT: int
AUTH_MODE_SYSDBA: int
AUTH_MODE_SYSOPER: int

# API Type Constants
STRING: type
NUMBER: type
DATETIME: type
BINARY: type
ROWID: type

Data Types and Constants

Exception Hierarchy

class Error(Exception): ...
class DatabaseError(Error): ...
class DataError(DatabaseError): ...
class IntegrityError(DatabaseError): ...
class InternalError(DatabaseError): ...
class NotSupportedError(DatabaseError): ...
class OperationalError(DatabaseError): ...
class ProgrammingError(DatabaseError): ...
class InterfaceError(Error): ...
class Warning(Exception): ...

Utility Functions

def makedsn(host, port, sid=None, service_name=None, region=None, sharding_key=None, super_sharding_key=None) -> str: ...
def enable_thin_mode(thin=True) -> None: ...
def is_thin_mode() -> bool: ...
def init_oracle_client(lib_dir=None, config_dir=None, error_url=None, driver_name=None) -> None: ...
def clientversion() -> tuple: ...

# Arrow Integration
def from_arrow(arrow_table, **kwargs): ...

# Hook Registration
def register_params_hook(hook_func) -> None: ...
def unregister_params_hook(hook_func) -> None: ...
def register_password_type(password_type) -> None: ...
def register_protocol(protocol_name, protocol_func) -> None: ...

# Constructor Functions (DB API 2.0 compatibility)
def Binary(data) -> bytes: ...
def Date(year, month, day) -> datetime.date: ...
def DateFromTicks(ticks) -> datetime.date: ...
def Time(hour, minute, second) -> datetime.time: ...
def TimeFromTicks(ticks) -> datetime.time: ...
def Timestamp(year, month, day, hour, minute, second) -> datetime.datetime: ...
def TimestampFromTicks(ticks) -> datetime.datetime: ...

Configuration

class ConnectParams:
    user: str
    password: str
    dsn: str
    # ... many other connection parameters

class PoolParams:
    min: int
    max: int
    increment: int
    # ... many other pool parameters

# Global defaults
defaults: object