tessl/pypi-cx-oracle

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

—

Pending

Overview

Eval results

Files

Database Connections

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

Core connection management for Oracle Database access including direct connections, connection strings, authentication modes, and connection properties.

Capabilities

Creating Connections

Connect to Oracle Database using various connection methods and authentication modes.

def connect(user=None, password=None, dsn=None, mode=DEFAULT_AUTH, handle=0, 
           pool=None, params=None, threaded=False, events=False, 
           purity=ATTR_PURITY_DEFAULT, newpassword=None, encoding=None, 
           nencoding=None, edition=None, appcontext=[], tag=None, 
           matchanytag=False, config_dir=None, driver_name=None, 
           shardingkey=[], supershardingkey=[], debug_jdwp=None, 
           connection_id_prefix=None, ssl_context=None, sdu=None, 
           pool_boundary=None, use_tcp_fast_open=None, ssl_version=None, 
           wallet_location=None, wallet_password=None, access_token=None, 
           external_handle=None, mode_from_py=None) -> Connection:
    """
    Create connection to Oracle Database.
    
    Parameters:
    - user (str): Username
    - password (str): Password  
    - dsn (str): Data source name/connect string
    - mode (int): Authentication mode (DEFAULT_AUTH, SYSDBA, SYSOPER, etc.)
    - threaded (bool): Enable thread safety
    - events (bool): Enable Oracle events
    - encoding (str): Character encoding (default: UTF-8)
    - nencoding (str): National character encoding
    - edition (str): Database edition
    - tag (str): Connection tag for pooling
    - config_dir (str): Oracle client configuration directory
    - ssl_context: SSL context for encrypted connections
    
    Returns:
    Connection object
    """

Usage examples:

# Basic connection
conn = cx_Oracle.connect("scott", "tiger", "localhost:1521/XE")

# Connection with specific encoding
conn = cx_Oracle.connect("user", "pass", "dsn", encoding="UTF-8")

# Administrative connection
conn = cx_Oracle.connect("sys", "password", "dsn", mode=cx_Oracle.SYSDBA)

# Connection with SSL
import ssl
ssl_context = ssl.create_default_context()
conn = cx_Oracle.connect("user", "pass", "dsn", ssl_context=ssl_context)

DSN Creation

Create properly formatted Data Source Names for Oracle connections.

def makedsn(host: str, port: int, sid=None, service_name=None, region=None, 
           sharding_key=None, super_sharding_key=None) -> str:
    """
    Create Oracle DSN from connection components.
    
    Parameters:
    - host (str): Database server hostname or IP
    - port (int): Database server port (typically 1521)
    - sid (str): Oracle SID (deprecated, use service_name)
    - service_name (str): Oracle service name
    - region (str): Oracle Cloud region
    - sharding_key (list): Sharding key for sharded databases
    - super_sharding_key (list): Super sharding key
    
    Returns:
    Formatted DSN string
    """

Usage examples:

# Service name (recommended)
dsn = cx_Oracle.makedsn("myhost", 1521, service_name="XEPDB1")

# Legacy SID format
dsn = cx_Oracle.makedsn("myhost", 1521, sid="XE")

# With sharding (Oracle 12.2+)
dsn = cx_Oracle.makedsn("myhost", 1521, service_name="service", 
                       sharding_key=[100], super_sharding_key=["region1"])

Client Library Management

Initialize and manage Oracle client library configuration.

def init_oracle_client(lib_dir=None, config_dir=None, error_url=None, 
                      driver_name=None) -> None:
    """
    Initialize Oracle client library with custom configuration.
    
    Parameters:
    - lib_dir (str): Oracle client library directory path
    - config_dir (str): Oracle client configuration directory  
    - error_url (str): URL for additional error information
    - driver_name (str): Custom driver name
    
    Raises:
    InterfaceError: If initialization fails
    """

def clientversion() -> tuple:
    """
    Get Oracle client library version information.
    
    Returns:
    5-tuple: (version, release, update, port_release, port_update)
    """

Usage examples:

# Initialize with custom library path
cx_Oracle.init_oracle_client(lib_dir="/opt/oracle/instantclient_19_8")

# Get client version
version = cx_Oracle.clientversion()
print(f"Client version: {version[0]}.{version[1]}.{version[2]}")

Connection Class

Represents an active connection to Oracle Database with methods for transaction control and resource management.

Connection Properties

class Connection:
    @property
    def autocommit(self) -> bool:
        """Auto-commit mode flag"""
        
    @property  
    def username(self) -> str:
        """Connected username"""
        
    @property
    def dsn(self) -> str:
        """Data source name"""
        
    @property
    def tnsentry(self) -> str:
        """TNS entry (alias for dsn)"""
        
    @property
    def version(self) -> str:
        """Database version string"""
        
    @property
    def encoding(self) -> str:
        """Character encoding"""
        
    @property
    def nencoding(self) -> str:
        """National character encoding"""
        
    @property
    def maxBytesPerCharacter(self) -> int:
        """Maximum bytes per character"""
        
    @property
    def current_schema(self) -> str:
        """Current schema name"""
        
    @property
    def edition(self) -> str:
        """Database edition"""
        
    @property
    def ltxid(self) -> bytes:
        """Logical transaction ID"""
        
    @property
    def stmtcachesize(self) -> int:
        """Statement cache size"""
        
    @property
    def tag(self) -> str:
        """Connection tag"""
        
    @property
    def call_timeout(self) -> int:
        """Call timeout in milliseconds"""
        
    @property
    def client_identifier(self) -> str:
        """Client identifier for monitoring"""
        
    @property
    def clientinfo(self) -> str:
        """Client info string"""
        
    @property
    def module(self) -> str:
        """Module name for monitoring"""
        
    @property
    def action(self) -> str:
        """Action name for monitoring"""
        
    @property
    def dbop(self) -> str:
        """Database operation name"""
        
    @property
    def inputtypehandler(self):
        """Input type handler function"""
        
    @property
    def outputtypehandler(self):
        """Output type handler function"""

Connection Methods

class Connection:
    def cursor(self) -> Cursor:
        """Create new cursor for executing SQL statements"""
        
    def commit(self) -> None:
        """Commit current transaction"""
        
    def rollback(self) -> None:
        """Rollback current transaction"""
        
    def begin(self, formatId=None, transactionId=None, branchId=None) -> None:
        """Begin new transaction (optionally distributed transaction)"""
        
    def prepare(self) -> bool:
        """Prepare transaction for two-phase commit"""
        
    def close(self) -> None:
        """Close connection and free resources"""
        
    def ping(self) -> None:
        """Test connection liveness"""
        
    def cancel(self) -> None:
        """Cancel currently executing operation"""
        
    def changepassword(self, oldpassword: str, newpassword: str) -> None:
        """Change user password"""
        
    def createlob(self, lobtype) -> LOB:
        """Create temporary LOB object"""
        
    def gettype(self, name: str) -> ObjectType:
        """Get Oracle object type by name"""
        
    def getSodaDatabase(self) -> SodaDatabase:
        """Get SODA database object for document operations"""
        
    def deqoptions(self) -> DeqOptions:
        """Create dequeue options object"""
        
    def enqoptions(self) -> EnqOptions:
        """Create enqueue options object"""
        
    def msgproperties(self, **kwargs) -> MessageProperties:
        """Create message properties object"""
        
    def deq(self, **kwargs) -> MessageProperties:
        """Dequeue message"""
        
    def enq(self, **kwargs) -> None:
        """Enqueue message"""
        
    def queue(self, name: str, **kwargs) -> Queue:
        """Get or create queue object"""
        
    def shutdown(self, mode=None, **kwargs) -> None:
        """Shutdown database (requires DBA privileges)"""
        
    def startup(self, force=False, restrict=False, pfile=None, **kwargs) -> None:
        """Startup database (requires DBA privileges)"""
        
    def __enter__(self) -> Connection:
        """Context manager entry"""
        
    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        """Context manager exit with automatic cleanup"""

Transaction Management

class Connection:
    def tpc_begin(self, xid) -> None:
        """Begin distributed transaction"""
        
    def tpc_prepare(self) -> None:
        """Prepare distributed transaction"""
        
    def tpc_commit(self, xid=None) -> None:
        """Commit distributed transaction"""
        
    def tpc_rollback(self, xid=None) -> None:
        """Rollback distributed transaction"""
        
    def tpc_recover(self) -> list:
        """Get list of pending distributed transactions"""

Subscription Management

class Connection:
    def subscribe(self, callback, sql=None, operations=None, qos=None, 
                 timeout=0, **kwargs) -> Subscription:
        """
        Create subscription for database change notifications.
        
        Parameters:
        - callback: Function to call when events occur
        - sql (str): SQL statement to monitor
        - operations (int): Operations to monitor (OPCODE_* constants)
        - qos (int): Quality of service flags
        - timeout (int): Subscription timeout in seconds
        
        Returns:
        Subscription object
        """
        
    def unsubscribe(self, subscription: Subscription) -> None:
        """Remove database change notification subscription"""

Authentication Modes

DEFAULT_AUTH: int       # Default authentication
SYSASM: int            # SYSASM administrative privilege  
SYSBKP: int            # SYSBKP administrative privilege
SYSDBA: int            # SYSDBA administrative privilege
SYSDGD: int            # SYSDGD administrative privilege  
SYSKMT: int            # SYSKMT administrative privilege
SYSOPER: int           # SYSOPER administrative privilege
SYSRAC: int            # SYSRAC administrative privilege
PRELIM_AUTH: int       # Preliminary authentication for startup/shutdown

Connection Context Management

cx_Oracle connections support Python's context manager protocol:

# Automatic connection cleanup
with cx_Oracle.connect("user", "pass", "dsn") as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM employees")
    # Connection automatically closed when exiting context

Error Handling

Connection-related errors raise specific exception types:

InterfaceError: Connection interface problems
DatabaseError: Database connection issues
OperationalError: Database operational problems

try:
    conn = cx_Oracle.connect("user", "wrongpass", "dsn")
except cx_Oracle.DatabaseError as e:
    error_obj, = e.args
    print(f"Oracle error {error_obj.code}: {error_obj.message}")

Install with Tessl CLI