Connect to an Oracle-compatible tenant of OceanBase Cloud by using the Python cx_Oracle driver

python3 --version

# Uninstall the old version (if any)
rpm -e libobclient
rpm -e obci

# Install the new version
rpm -ivh libobclient-2.2.11-42025062010.el7.x86_64.rpm
rpm -ivh obci-2.1.1-342025070917.el7.x86_64.rpm

# Switch to the directory where the driver is stored
cd /directory/of/cx_Oracle-8.3.0.tar.gz

# Decompress the cx_Oracle driver package
tar -xvf cx_Oracle-8.3.0.tar.gz

# Go to the decompressed directory
cd cx_Oracle-8.3.0

# Install the cx_Oracle driver
python3 setup.py install

export LD_LIBRARY_PATH=/u01/obclient/lib/:$LD_LIBRARY_PATH

import cx_Oracle

# Database connection information
username = 'oracle001'
password = '*****'
oracle_connection = 't5******.********.oceanbase.cloud:1521/sys'

# Create a database connection
conn = cx_Oracle.connect(username, password, oracle_connection)

def exec_sql(sql):
    """Execute an SQL statement."""
    cur = conn.cursor()
    try:
        cur.execute(sql)
        conn.commit()
        print(f"SQL executed successfully: {sql}")
    except Exception as e:
        print(f"SQL execution failed: {sql}")
        print(f"Error message: {e}")
    finally:
        cur.close()

def print_data(sql):
    """Query and print data."""
    cur = conn.cursor()
    try:
        cur.execute(sql)
        data = cur.fetchall()
        print(f"Query result: {data}")
        return data
    except Exception as e:
        print(f"Query failed: {sql}")
        print(f"Error message: {e}")
        return None
    finally:
        cur.close()

def test_char_types():
    """Test character types."""
    print("\n=== Test character types ===")
    exec_sql("DROP TABLE test_char")
    exec_sql("""
        CREATE TABLE test_char (
            id INT,
            a VARCHAR2(20),
            b CHAR(10),
            c NCHAR(10),
            d NVARCHAR2(10)
        )
    """)
    exec_sql("INSERT INTO test_char VALUES (1, 'hello', 'adffdf', '2df4d', 'dsf44f')")
    print_data("SELECT * FROM test_char")

def test_number_types():
    """Test numeric types."""
    print("\n=== Test numeric types ===")
    exec_sql("DROP TABLE test_number")
    exec_sql("""
        CREATE TABLE test_number (
            a NUMBER,
            b FLOAT(126),
            c BINARY_FLOAT,
            d BINARY_DOUBLE
        )
    """)
    exec_sql("INSERT INTO test_number VALUES (12.32, 12.34, 14.23, 123.3433)")
    print_data("SELECT * FROM test_number")

def test_time_types():
    """Test time types."""
    print("\n=== Test time types ===")
    exec_sql("DROP TABLE test_time")
    exec_sql("""
        CREATE TABLE test_time (
            a DATE,
            b TIMESTAMP,
            c TIMESTAMP WITH TIME ZONE,
            d TIMESTAMP WITH LOCAL TIME ZONE
        )
    """)
    exec_sql("""
        INSERT INTO test_time VALUES (
            TIMESTAMP'2022-08-29 14:44:30',
            TIMESTAMP'2022-08-29 14:44:30',
            TIMESTAMP'2022-08-29 14:44:30',
            TIMESTAMP'2022-08-29 14:44:30'
        )
    """)
    print_data("SELECT * FROM test_time")

def test_lob_types():
    """Test LOB types."""
    print("\n=== Test LOB types ===")
    exec_sql("DROP TABLE test_lob")
    exec_sql("""
        CREATE TABLE test_lob (
            a CLOB,
            b BLOB,
            c RAW(100)
        )
    """)
    exec_sql("INSERT INTO test_lob VALUES ('sdfdslkfjldsf', '31323334353637', '31323334')")
    print_data("SELECT * FROM test_lob")

def main():
    """Main function."""
    try:
        print("Start testing the connection to an Oracle database in OceanBase Cloud...")

        # Test various data types
        test_char_types()
        test_number_types()
        test_time_types()
        test_lob_types()

        print("\nAll tests completed!")

    except Exception as e:
        print(f"An error occurred during testing: {e}")
    finally:
        # Close the database connection
        if conn:
            conn.close()
            print("Database connection closed")

if __name__ == "__main__":
    main()

python3 test_oracle.py

import cx_Oracle
from cx_Oracle import SessionPool

def test_simple_pool():
    """Connection pool test."""
    try:
        print("Creating a connection pool...")
        # Create a connection pool.
        pool = SessionPool(
            user="oracle001",
            password="******",
            dsn="t5******.********.oceanbase.cloud:1521/sys",
            min=2,
            max=5,
            increment=1
        )
        print(f"Connection pool created.")
        print(f"- Minimum connections: {pool.min}")
        print(f"- Maximum connections: {pool.max}")
        print(f"- Increment: {pool.increment}")

        print("\nTesting connection pool usage...")
        # Acquire a connection from the connection pool.
        with pool.acquire() as connection:
            with connection.cursor() as cursor:
                # Execute a query.
                cursor.execute("SELECT COUNT(*) FROM test_char")
                result = cursor.fetchone()
                print(f"Number of records in the test_char table: {result[0]}")

                # Execute another query.
                cursor.execute("SELECT * FROM test_char WHERE id = 1")
                data = cursor.fetchone()
                print(f"Query result: {data}")

        print(f"\nConnection pool status:")
        print(f"- Number of open connections: {pool.opened}")
        print(f"- Number of busy connections: {pool.busy}")

        print("\nConnection pool test completed.")
        # Note: Do not call pool.close() to avoid a segmentation fault.

    except Exception as e:
        print(f"Error occurred: {e}")

if __name__ == "__main__":
    test_simple_pool()