Python interface to Oracle Database with thin and thick connectivity modes
—
Comprehensive support for all Oracle data types, authentication modes, and configuration constants. Oracle python-oracledb provides extensive type system coverage enabling seamless data exchange between Python and Oracle Database, including support for modern data types like JSON and vectors.
Constants representing all Oracle database types for use in variable declarations, type checking, and data conversion.
# Character and String Types
DB_TYPE_CHAR: type # Fixed-length character data
DB_TYPE_VARCHAR: type # Variable-length character data
DB_TYPE_NCHAR: type # Fixed-length national character data
DB_TYPE_NVARCHAR: type # Variable-length national character data
DB_TYPE_LONG: type # Long character data (deprecated)
DB_TYPE_LONG_NVARCHAR: type # Long national character data
# Numeric Types
DB_TYPE_NUMBER: type # Oracle NUMBER type
DB_TYPE_BINARY_INTEGER: type # Binary integer
DB_TYPE_BINARY_FLOAT: type # Binary float (32-bit)
DB_TYPE_BINARY_DOUBLE: type # Binary double (64-bit)
# Date and Time Types
DB_TYPE_DATE: type # Date and time
DB_TYPE_TIMESTAMP: type # Timestamp
DB_TYPE_TIMESTAMP_TZ: type # Timestamp with timezone
DB_TYPE_TIMESTAMP_LTZ: type # Timestamp with local timezone
DB_TYPE_INTERVAL_DS: type # Interval day to second
DB_TYPE_INTERVAL_YM: type # Interval year to month
# Large Object Types
DB_TYPE_BLOB: type # Binary Large Object
DB_TYPE_CLOB: type # Character Large Object
DB_TYPE_NCLOB: type # National Character Large Object
DB_TYPE_BFILE: type # Binary File (external)
# Other Types
DB_TYPE_RAW: type # Raw binary data
DB_TYPE_LONG_RAW: type # Long raw binary data
DB_TYPE_ROWID: type # Physical row identifier
DB_TYPE_UROWID: type # Universal row identifier
DB_TYPE_CURSOR: type # Cursor/REF CURSOR
DB_TYPE_OBJECT: type # User-defined object type
DB_TYPE_BOOLEAN: type # Boolean (Oracle 12c+)
DB_TYPE_JSON: type # JSON data type (Oracle 21c+)
DB_TYPE_VECTOR: type # Vector data type (Oracle 23c+)
DB_TYPE_XMLTYPE: type # XMLType
DB_TYPE_UNKNOWN: type # Unknown type
# Legacy Type Aliases (for backward compatibility)
BFILE: type # Alias for DB_TYPE_BFILE
BLOB: type # Alias for DB_TYPE_BLOB
BOOLEAN: type # Alias for DB_TYPE_BOOLEAN
CLOB: type # Alias for DB_TYPE_CLOB
CURSOR: type # Alias for DB_TYPE_CURSOR
FIXED_CHAR: type # Alias for DB_TYPE_CHAR
FIXED_NCHAR: type # Alias for DB_TYPE_NCHAR
INTERVAL: type # Alias for DB_TYPE_INTERVAL_DS
LONG_BINARY: type # Alias for DB_TYPE_LONG_RAW
LONG_STRING: type # Alias for DB_TYPE_LONG
NATIVE_INT: type # Alias for DB_TYPE_BINARY_INTEGER
NATIVE_FLOAT: type # Alias for DB_TYPE_BINARY_DOUBLE
NCHAR: type # Alias for DB_TYPE_NVARCHAR
NCLOB: type # Alias for DB_TYPE_NCLOB
OBJECT: type # Alias for DB_TYPE_OBJECT
TIMESTAMP: type # Alias for DB_TYPE_TIMESTAMPPython DB API 2.0 standard type constants for portable database programming.
# Standard DB API Types
STRING: type # String data type
BINARY: type # Binary data type
NUMBER: type # Numeric data type
DATETIME: type # Date/time data type
ROWID: type # Row identifier typeConstants for specifying database authentication modes and privileges.
# Authentication Modes
AUTH_MODE_DEFAULT: int # Default authentication
AUTH_MODE_SYSDBA: int # SYSDBA privilege
AUTH_MODE_SYSOPER: int # SYSOPER privilege
AUTH_MODE_SYSASM: int # SYSASM privilege (ASM)
AUTH_MODE_SYSBKP: int # SYSBKP privilege (backup)
AUTH_MODE_SYSDGD: int # SYSDGD privilege (Data Guard)
AUTH_MODE_SYSKMT: int # SYSKMT privilege (Key Management)
AUTH_MODE_SYSRAC: int # SYSRAC privilege (RAC)
AUTH_MODE_PRELIM: int # Preliminary authentication
# Legacy Authentication Aliases
DEFAULT_AUTH: int # Alias for AUTH_MODE_DEFAULT
SYSDBA: int # Alias for AUTH_MODE_SYSDBA
SYSOPER: int # Alias for AUTH_MODE_SYSOPER
SYSASM: int # Alias for AUTH_MODE_SYSASM
SYSBKP: int # Alias for AUTH_MODE_SYSBKP
SYSDGD: int # Alias for AUTH_MODE_SYSDGD
SYSKMT: int # Alias for AUTH_MODE_SYSKMT
SYSRAC: int # Alias for AUTH_MODE_SYSRAC
PRELIM_AUTH: int # Alias for AUTH_MODE_PRELIMConstants for configuring connection pool behavior and connection acquisition modes.
# Pool Get Modes
POOL_GETMODE_WAIT: int # Wait for available connection
POOL_GETMODE_NOWAIT: int # Return immediately if no connection
POOL_GETMODE_FORCEGET: int # Create connection beyond max limit
POOL_GETMODE_TIMEDWAIT: int # Wait with timeout
# Pool Purity Levels
PURITY_DEFAULT: int # Default purity
PURITY_NEW: int # New session required
PURITY_SELF: int # Self-contained session
# Legacy Pool Aliases
SPOOL_ATTRVAL_WAIT: int # Alias for POOL_GETMODE_WAIT
SPOOL_ATTRVAL_NOWAIT: int # Alias for POOL_GETMODE_NOWAIT
SPOOL_ATTRVAL_FORCEGET: int # Alias for POOL_GETMODE_FORCEGET
SPOOL_ATTRVAL_TIMEDWAIT: int # Alias for POOL_GETMODE_TIMEDWAIT
ATTR_PURITY_DEFAULT: int # Alias for PURITY_DEFAULT
ATTR_PURITY_NEW: int # Alias for PURITY_NEW
ATTR_PURITY_SELF: int # Alias for PURITY_SELFConstants for Oracle Advanced Queuing operations including message delivery, dequeue modes, and queue management.
# Message Delivery Modes
MSG_BUFFERED: int # Buffered messages
MSG_PERSISTENT: int # Persistent messages
MSG_PERSISTENT_OR_BUFFERED: int # Persistent or buffered
# Dequeue Modes
DEQ_BROWSE: int # Browse without removing
DEQ_LOCKED: int # Lock message
DEQ_REMOVE: int # Remove message
DEQ_REMOVE_NODATA: int # Remove without returning data
# Dequeue Navigation
DEQ_FIRST_MSG: int # First message
DEQ_NEXT_MSG: int # Next message
DEQ_NEXT_TRANSACTION: int # Next transaction
# Visibility Modes
DEQ_IMMEDIATE: int # Immediate visibility
DEQ_ON_COMMIT: int # Visible on commit
ENQ_IMMEDIATE: int # Immediate enqueue
ENQ_ON_COMMIT: int # Enqueue on commit
# Wait Modes
DEQ_NO_WAIT: int # Don't wait
DEQ_WAIT_FOREVER: int # Wait indefinitely
# Message States
MSG_EXPIRED: int # Message expired
MSG_PROCESSED: int # Message processed
MSG_READY: int # Message ready
MSG_WAITING: int # Message waiting
# Message Options
MSG_NO_DELAY: int # No delay
MSG_NO_EXPIRATION: int # No expirationConstants for database administration operations including shutdown modes and operation codes.
# Database Shutdown Modes
DBSHUTDOWN_ABORT: int # Abort shutdown
DBSHUTDOWN_FINAL: int # Final shutdown phase
DBSHUTDOWN_IMMEDIATE: int # Immediate shutdown
DBSHUTDOWN_TRANSACTIONAL: int # Transactional shutdown
DBSHUTDOWN_TRANSACTIONAL_LOCAL: int # Local transactional shutdown
# Database Operation Codes
OPCODE_ALLOPS: int # All operations
OPCODE_ALLROWS: int # All rows
OPCODE_INSERT: int # Insert operations
OPCODE_UPDATE: int # Update operations
OPCODE_DELETE: int # Delete operations
OPCODE_ALTER: int # Alter operations
OPCODE_DROP: int # Drop operations
# Event Types
EVENT_NONE: int # No event
EVENT_STARTUP: int # Database startup
EVENT_SHUTDOWN: int # Database shutdown
EVENT_SHUTDOWN_ANY: int # Any shutdown event
EVENT_DEREG: int # Deregistration event
EVENT_OBJCHANGE: int # Object change event
EVENT_QUERYCHANGE: int # Query change event
EVENT_AQ: int # Advanced Queuing eventConstants for distributed transaction management using two-phase commit protocol.
# TPC Begin Flags
TPC_BEGIN_JOIN: int # Join existing transaction
TPC_BEGIN_NEW: int # Start new transaction
TPC_BEGIN_PROMOTE: int # Promote to distributed
TPC_BEGIN_RESUME: int # Resume suspended transaction
# TPC End Flags
TPC_END_NORMAL: int # Normal end
TPC_END_SUSPEND: int # Suspend transactionConstants for Oracle 23c vector data type format specifications.
# Vector Formats
VECTOR_FORMAT_BINARY: int # Binary vector format
VECTOR_FORMAT_FLOAT32: int # 32-bit float vector format
VECTOR_FORMAT_FLOAT64: int # 64-bit float vector format
VECTOR_FORMAT_INT8: int # 8-bit integer vector formatPython Database API 2.0 mandated constants providing metadata about the driver capabilities.
# DB API Metadata
apilevel: str # API level ("2.0")
threadsafety: int # Thread safety level (2)
paramstyle: str # Parameter style ("named")import oracledb
from datetime import date, datetime
from decimal import Decimal
connection = oracledb.connect(user="hr", password="password", dsn="localhost/xepdb1")
# Create table with various data types
with connection.cursor() as cursor:
cursor.execute("""
CREATE TABLE data_types_demo (
id NUMBER PRIMARY KEY,
text_data VARCHAR2(100),
number_data NUMBER(10,2),
date_data DATE,
timestamp_data TIMESTAMP,
boolean_data NUMBER(1) CHECK (boolean_data IN (0,1)),
raw_data RAW(50),
clob_data CLOB,
blob_data BLOB
)
""")
# Insert data with proper type handling
with connection.cursor() as cursor:
cursor.execute("""
INSERT INTO data_types_demo (
id, text_data, number_data, date_data, timestamp_data,
boolean_data, raw_data, clob_data, blob_data
) VALUES (
:1, :2, :3, :4, :5, :6, :7, :8, :9
)
""", [
1, # NUMBER
"Sample text", # VARCHAR2
Decimal('123.45'), # NUMBER with precision
date(2024, 1, 15), # DATE
datetime(2024, 1, 15, 10, 30, 45), # TIMESTAMP
1, # Boolean as NUMBER
b'\x01\x02\x03\x04', # RAW
"Large text content", # CLOB
b'Binary data content' # BLOB
])
connection.commit()
# Query with type introspection
with connection.cursor() as cursor:
cursor.execute("SELECT * FROM data_types_demo WHERE id = 1")
# Examine column metadata
print("Column Information:")
for i, desc in enumerate(cursor.description):
print(f" Column {i}: {desc[0]} - Type: {desc[1]}")
# Fetch and display data
row = cursor.fetchone()
print(f"\nRetrieved Data:")
for i, value in enumerate(row):
column_name = cursor.description[i][0]
print(f" {column_name}: {value} ({type(value).__name__})")
connection.close()import oracledb
connection = oracledb.connect(user="hr", password="password", dsn="localhost/xepdb1")
# Using type constants for variable creation
with connection.cursor() as cursor:
# Create variables with specific types
string_var = cursor.var(oracledb.DB_TYPE_VARCHAR, size=100)
number_var = cursor.var(oracledb.DB_TYPE_NUMBER)
date_var = cursor.var(oracledb.DB_TYPE_DATE)
clob_var = cursor.var(oracledb.DB_TYPE_CLOB)
# Use in stored procedure call
cursor.callproc("some_procedure", [string_var, number_var, date_var, clob_var])
# Retrieve values with proper types
result_string = string_var.getvalue()
result_number = number_var.getvalue()
result_date = date_var.getvalue()
result_clob = clob_var.getvalue()
# Type checking in application logic
def process_oracle_value(value, oracle_type):
"""Process value based on Oracle type."""
if oracle_type == oracledb.DB_TYPE_VARCHAR:
return str(value) if value is not None else ""
elif oracle_type == oracledb.DB_TYPE_NUMBER:
return float(value) if value is not None else 0.0
elif oracle_type == oracledb.DB_TYPE_DATE:
return value.strftime('%Y-%m-%d') if value else None
elif oracle_type in (oracledb.DB_TYPE_CLOB, oracledb.DB_TYPE_BLOB):
return value.read() if value else None
else:
return value
connection.close()import oracledb
# Connect with different authentication modes
try:
# Regular connection
regular_conn = oracledb.connect(
user="hr",
password="password",
dsn="localhost/xepdb1",
mode=oracledb.AUTH_MODE_DEFAULT
)
print("Regular connection successful")
regular_conn.close()
# SYSDBA connection (requires DBA privileges)
dba_conn = oracledb.connect(
user="sys",
password="system_password",
dsn="localhost/xepdb1",
mode=oracledb.AUTH_MODE_SYSDBA
)
print("SYSDBA connection successful")
# Check connection privileges
with dba_conn.cursor() as cursor:
cursor.execute("SELECT USER FROM DUAL")
user = cursor.fetchone()[0]
print(f"Connected as: {user}")
dba_conn.close()
except oracledb.DatabaseError as e:
print(f"Connection failed: {e}")import oracledb
# Create pool with specific configuration
pool = oracledb.create_pool(
user="hr",
password="password",
dsn="localhost/xepdb1",
min=5,
max=20,
increment=3,
getmode=oracledb.POOL_GETMODE_WAIT, # Wait for available connection
homogeneous=True,
timeout=300 # 5 minutes
)
# Acquire connection with specific purity
connection = pool.acquire(
cclass="OLTP",
purity=oracledb.PURITY_SELF # Self-contained session
)
print(f"Pool status: {pool.busy}/{pool.opened} connections in use")
# Use connection
with connection.cursor() as cursor:
cursor.execute("SELECT COUNT(*) FROM employees")
count = cursor.fetchone()[0]
print(f"Employee count: {count}")
# Release back to pool
pool.release(connection)
# Try different get mode
try:
quick_conn = pool.acquire(
getmode=oracledb.POOL_GETMODE_NOWAIT # Don't wait
)
print("Got connection immediately")
pool.release(quick_conn)
except oracledb.DatabaseError as e:
print(f"No connection available: {e}")
pool.close()import oracledb
import json
connection = oracledb.connect(user="hr", password="password", dsn="localhost/xepdb1")
# Working with JSON data type (Oracle 21c+)
with connection.cursor() as cursor:
try:
cursor.execute("""
CREATE TABLE json_demo (
id NUMBER PRIMARY KEY,
json_data JSON
)
""")
# Insert JSON data
json_content = {
"name": "John Doe",
"age": 30,
"department": "Engineering",
"skills": ["Python", "Oracle", "SQL"]
}
cursor.execute("""
INSERT INTO json_demo (id, json_data)
VALUES (:1, :2)
""", [1, json.dumps(json_content)])
connection.commit()
# Query JSON data
cursor.execute("SELECT json_data FROM json_demo WHERE id = 1")
result = cursor.fetchone()
retrieved_json = json.loads(result[0])
print(f"Retrieved JSON: {retrieved_json}")
except oracledb.DatabaseError as e:
if "ORA-00902" in str(e): # Invalid datatype
print("JSON datatype not available (requires Oracle 21c+)")
else:
print(f"JSON demo error: {e}")
# Working with Vector data type (Oracle 23c+)
with connection.cursor() as cursor:
try:
cursor.execute("""
CREATE TABLE vector_demo (
id NUMBER PRIMARY KEY,
embedding VECTOR(128, FLOAT32)
)
""")
# Vector operations would go here
print("Vector table created successfully")
except oracledb.DatabaseError as e:
if "ORA-00902" in str(e):
print("Vector datatype not available (requires Oracle 23c+)")
else:
print(f"Vector demo error: {e}")
connection.close()import oracledb
def analyze_table_types(connection, table_name):
"""Analyze data types in a table."""
with connection.cursor() as cursor:
cursor.execute(f"SELECT * FROM {table_name} WHERE ROWNUM <= 1")
cursor.fetchall() # Consume results to get metadata
print(f"Table: {table_name}")
print("Column Type Analysis:")
for i, desc in enumerate(cursor.description):
column_name = desc[0]
oracle_type = desc[1]
display_size = desc[2]
internal_size = desc[3]
precision = desc[4]
scale = desc[5]
null_ok = desc[6]
# Map Oracle type to readable name
type_names = {
oracledb.DB_TYPE_VARCHAR: "VARCHAR2",
oracledb.DB_TYPE_CHAR: "CHAR",
oracledb.DB_TYPE_NUMBER: "NUMBER",
oracledb.DB_TYPE_DATE: "DATE",
oracledb.DB_TYPE_TIMESTAMP: "TIMESTAMP",
oracledb.DB_TYPE_CLOB: "CLOB",
oracledb.DB_TYPE_BLOB: "BLOB",
oracledb.DB_TYPE_RAW: "RAW"
}
type_name = type_names.get(oracle_type, f"Unknown({oracle_type})")
nullable = "NULL" if null_ok else "NOT NULL"
size_info = ""
if precision:
if scale:
size_info = f"({precision},{scale})"
else:
size_info = f"({precision})"
elif display_size:
size_info = f"({display_size})"
print(f" {column_name}: {type_name}{size_info} {nullable}")
connection = oracledb.connect(user="hr", password="password", dsn="localhost/xepdb1")
# Analyze built-in tables
analyze_table_types(connection, "employees")
print()
analyze_table_types(connection, "departments")
connection.close()Install with Tessl CLI
npx tessl i tessl/pypi-oracledb