tessl/pypi-oracledb
Python interface to Oracle Database with thin and thick connectivity modes
—
Pending
connectivity.mddocs/
Database Connectivity
Core connection management including single connections, connection pools, and connection parameter configuration. Oracle python-oracledb supports both synchronous and asynchronous operations with thin mode (direct connection) and thick mode (using Oracle Client libraries) connectivity options.
Capabilities
Creating Connections
Create database connections with various authentication and configuration options.
def connect(
dsn: Optional[str] = None,
*,
pool: Optional[ConnectionPool] = None,
pool_alias: Optional[str] = None,
conn_class: Type[Connection] = Connection,
params: Optional[ConnectParams] = None,
user: Optional[str] = None,
proxy_user: Optional[str] = None,
password: Optional[str] = None,
newpassword: Optional[str] = None,
wallet_password: Optional[str] = None,
access_token: Optional[Union[str, tuple, Callable]] = None,
host: Optional[str] = None,
port: Optional[int] = None,
protocol: Optional[str] = None,
https_proxy: Optional[str] = None,
https_proxy_port: Optional[int] = None,
service_name: Optional[str] = None,
instance_name: Optional[str] = None,
sid: Optional[str] = None,
server_type: Optional[str] = None,
cclass: Optional[str] = None,
purity: Optional[Purity] = None,
expire_time: Optional[int] = None,
retry_count: Optional[int] = None,
retry_delay: Optional[int] = None,
tcp_connect_timeout: Optional[float] = None,
ssl_server_dn_match: Optional[bool] = None,
ssl_server_cert_dn: Optional[str] = None,
wallet_location: Optional[str] = None,
events: Optional[bool] = None,
externalauth: Optional[bool] = None,
mode: Optional[AuthMode] = None,
disable_oob: Optional[bool] = None,
stmtcachesize: Optional[int] = None,
edition: Optional[str] = None,
tag: Optional[str] = None,
matchanytag: Optional[bool] = None,
config_dir: Optional[str] = None,
appcontext: Optional[list] = None,
shardingkey: Optional[list] = None,
supershardingkey: Optional[list] = None,
debug_jdwp: Optional[str] = None,
connection_id_prefix: Optional[str] = None,
ssl_context: Optional[Any] = None,
sdu: Optional[int] = None,
pool_boundary: Optional[str] = None,
use_tcp_fast_open: Optional[bool] = None,
ssl_version: Optional[Any] = None,
program: Optional[str] = None,
machine: Optional[str] = None,
terminal: Optional[str] = None,
osuser: Optional[str] = None,
driver_name: Optional[str] = None,
use_sni: Optional[bool] = None,
thick_mode_dsn_passthrough: Optional[bool] = None,
extra_auth_params: Optional[dict] = None,
pool_name: Optional[str] = None,
handle: Optional[int] = None,
) -> Connection:
"""
Creates a connection to an Oracle database and returns a Connection object.
Parameters:
- dsn (str): Data source name, connection string, or Easy Connect string
- pool (ConnectionPool): Connection pool to acquire connection from
- pool_alias (str): Named pool alias to acquire connection from
- params (ConnectParams): Pre-configured connection parameters object
- user (str): Username for database authentication
- proxy_user (str): Proxy user name for proxy authentication
- password (str): Password for database authentication
- newpassword (str): New password for password change on connect
- wallet_password (str): Password for encrypted wallet (thin mode)
- access_token (Union[str, tuple, Callable]): OAuth2 or OCI IAM token
- host (str): Database server hostname or IP address
- port (int): Database listener port number (default: 1521)
- protocol (str): Connection protocol ("tcp" or "tcps")
- service_name (str): Oracle service name
- sid (str): Oracle System Identifier (SID)
- mode (AuthMode): Authentication mode (SYSDBA, SYSOPER, etc.)
- Other parameters: Various connection and SSL configuration options
Returns:
Connection: Database connection object
"""
def connect_async(
dsn: Optional[str] = None,
*,
pool: Optional[AsyncConnectionPool] = None,
pool_alias: Optional[str] = None,
conn_class: Type[AsyncConnection] = AsyncConnection,
params: Optional[ConnectParams] = None,
**kwargs # Same parameters as connect()
) -> AsyncConnection:
"""
Creates an asynchronous connection to an Oracle database.
Parameters: Same as connect() but returns AsyncConnection
Returns:
AsyncConnection: Asynchronous database connection object
"""Connection Class
Manage database connections with transaction control, cursor creation, and administrative operations.
class Connection:
"""Synchronous database connection."""
# Properties
autocommit: bool
call_timeout: int
current_schema: str
dsn: str
username: str
version: tuple
transaction_in_progress: bool
thin: bool
warning: Warning
def cursor(self, scrollable=False) -> Cursor:
"""
Create a cursor for executing SQL statements.
Parameters:
- scrollable (bool): Enable scrollable cursor
Returns:
Cursor object
"""
def commit(self) -> None:
"""Commit the current transaction."""
def rollback(self) -> None:
"""Rollback the current transaction."""
def close(self) -> None:
"""Close the connection and free resources."""
def ping(self) -> None:
"""Test if the connection is still active."""
def begin(self, formatId=None, transactionId=None, branchId=None) -> None:
"""
Begin a new transaction or join distributed transaction.
Parameters:
- formatId (int): Format identifier for distributed transaction
- transactionId (bytes): Transaction identifier
- branchId (bytes): Branch identifier
"""
def prepare(self) -> bool:
"""
Prepare the current transaction for two-phase commit.
Returns:
bool: True if transaction was prepared, False if no work to prepare
"""
def subscribe(
self,
namespace=SUBSCR_NAMESPACE_DBCHANGE,
protocol=SUBSCR_PROTO_CALLBACK,
callback=None,
timeout=0,
operations=OPCODE_ALLOPS,
port=0,
qos=0,
ip_address=None,
grouping_class=SUBSCR_GROUPING_CLASS_NONE,
grouping_value=0,
grouping_type=SUBSCR_GROUPING_TYPE_SUMMARY,
name=None,
client_initiated=False,
recipient_name=None
) -> Subscription:
"""
Create a subscription for database event notifications.
Parameters:
- namespace (int): Subscription namespace
- protocol (int): Notification protocol
- callback: Callback function for notifications
- timeout (int): Subscription timeout
Returns:
Subscription object
"""
def createlob(self, lob_type) -> LOB:
"""
Create a temporary LOB.
Parameters:
- lob_type: LOB type (DB_TYPE_BLOB, DB_TYPE_CLOB, etc.)
Returns:
LOB object
"""
def gettype(self, name) -> DbObjectType:
"""
Get database object type by name.
Parameters:
- name (str): Object type name
Returns:
DbObjectType object
"""
def cancel(self) -> None:
"""Cancel long-running database operation."""
def shutdown(self, mode=DBSHUTDOWN_IMMEDIATE) -> None:
"""
Shutdown the database.
Parameters:
- mode (int): Shutdown mode
"""
def startup(self, force=False, restrict=False, pfile=None, spfile=None) -> None:
"""
Startup the database.
Parameters:
- force (bool): Force startup
- restrict (bool): Restrict mode
- pfile (str): Parameter file path
- spfile (str): Server parameter file path
"""Async Connection Class
Asynchronous version of Connection with async/await support.
class AsyncConnection:
"""Asynchronous database connection."""
# Properties (same as Connection)
autocommit: bool
call_timeout: int
current_schema: str
dsn: str
username: str
version: tuple
transaction_in_progress: bool
thin: bool
warning: Warning
async def cursor(self, scrollable=False) -> AsyncCursor:
"""
Create an async cursor for executing SQL statements.
Parameters:
- scrollable (bool): Enable scrollable cursor
Returns:
AsyncCursor object
"""
async def commit(self) -> None:
"""Commit the current transaction."""
async def rollback(self) -> None:
"""Rollback the current transaction."""
async def close(self) -> None:
"""Close the connection and free resources."""
async def ping(self) -> None:
"""Test if the connection is still active."""Connection Parameters
Configure connection parameters using ConnectParams class.
class ConnectParams:
"""Connection parameter configuration."""
user: str
password: str
dsn: str
mode: int
encoding: str
nencoding: str
edition: str
appcontext: list
tag: str
matchanytag: bool
config_dir: str
appname: str
stmtcachesize: int
disable_oob: bool
auth_token: str
access_token: str
expires_in: int
private_key: str
wallet_location: str
wallet_password: str
cclass: str
purity: int
server_type: str
ctype: str
sdu: int
pool_boundary: str
use_tcp_fast_open: bool
ssl_server_dn_match: bool
ssl_server_cert_dn: str
ssl_version: str
https_proxy: str
https_proxy_port: int
debug_jdwp: str
connection_id_prefix: str
ssl_context: object
soda_metadata_cache: bool
fetch_lobs: bool
fetch_decimals: bool
fetch_info: list
autocommit: bool
call_timeout: int
retry_count: int
retry_delay: int
externalauth: bool
homogeneous: boolDSN Construction
Create data source names (DSN) for database connections.
def makedsn(
host,
port,
sid=None,
service_name=None,
region=None,
sharding_key=None,
super_sharding_key=None,
cclass=None,
purity=None,
expire_time=None,
retry_count=None,
retry_delay=None,
tcp_connect_timeout=None,
ssl_server_dn_match=None,
ssl_server_cert_dn=None,
wallet_location=None,
**kwargs
) -> str:
"""
Create a data source name string.
Parameters:
- host (str): Database server hostname or IP
- port (int): Database server port number
- sid (str): Oracle System Identifier (exclusive with service_name)
- service_name (str): Service name (exclusive with sid)
- region (str): Oracle Cloud region
- sharding_key (list): Sharding key values
- super_sharding_key (list): Super sharding key values
Returns:
str: Formatted DSN string
"""Usage Examples
Basic Connection
import oracledb
import getpass
# Simple connection
username = "hr"
dsn = "localhost:1521/xepdb1"
password = getpass.getpass(f"Password for {username}: ")
connection = oracledb.connect(user=username, password=password, dsn=dsn)
print(f"Connected to Oracle Database version: {connection.version}")
connection.close()Connection with Advanced Options
import oracledb
# Connection with advanced configuration
connection = oracledb.connect(
user="scott",
password="tiger",
dsn="myhost:1521/myservice",
encoding="UTF-8",
autocommit=False,
call_timeout=30000, # 30 seconds
appname="MyApplication",
stmtcachesize=50
)
# Use connection
with connection.cursor() as cursor:
cursor.execute("SELECT COUNT(*) FROM employees")
count = cursor.fetchone()[0]
print(f"Employee count: {count}")
connection.close()Async Connection
import asyncio
import oracledb
async def main():
# Create async connection
connection = await oracledb.connect_async(
user="hr",
password="password",
dsn="localhost/xepdb1"
)
async with connection.cursor() as cursor:
await cursor.execute("SELECT sysdate FROM dual")
result = await cursor.fetchone()
print(f"Current date: {result[0]}")
await connection.close()
# Run async function
asyncio.run(main())Install with Tessl CLI
npx tessl i tessl/pypi-oracledb