Python 驱动连接 OceanBase 数据库 Oracle 模式指南

python3 --version

# 卸载旧版本（如果存在）
rpm -e libobclient
rpm -e obci

# 安装新版本
rpm -ivh libobclient-2.2.11-42025062010.el7.x86_64.rpm
rpm -ivh obci-2.1.1-342025070917.el7.x86_64.rpm

# 切换到驱动安装目录
cd /u01/obclient/python

# 解压 cx_Oracle 驱动包
tar -xvf cx_Oracle-8.3.0.tar.gz

# 进入解压后的目录
cd cx_Oracle-8.3.0

# 安装 cx_Oracle 驱动
python3 setup.py install

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

import cx_Oracle

# 数据库连接信息
username = 'test@oracle'
password = 'test'
oracle_connection = 'xxx.xxx.xxx.xxx:2881/TEST'

# 创建数据库连接
conn = cx_Oracle.connect(username, password, oracle_connection)

def exec_sql(sql):
    """执行 SQL 语句"""
    cur = conn.cursor()
    try:
        cur.execute(sql)
        conn.commit()
        print(f"SQL 执行成功: {sql}")
    except Exception as e:
        print(f"SQL 执行失败: {sql}")
        print(f"错误信息: {e}")
    finally:
        cur.close()

def print_data(sql):
    """查询并打印数据"""
    cur = conn.cursor()
    try:
        cur.execute(sql)
        data = cur.fetchall()
        print(f"查询结果: {data}")
        return data
    except Exception as e:
        print(f"查询失败: {sql}")
        print(f"错误信息: {e}")
        return None
    finally:
        cur.close()

def test_char_types():
    """测试字符类型"""
    print("\n=== 测试字符类型 ===")
    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():
    """测试数值类型"""
    print("\n=== 测试数值类型 ===")
    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():
    """测试时间类型"""
    print("\n=== 测试时间类型 ===")
    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():
    """测试 LOB 类型"""
    print("\n=== 测试 LOB 类型 ===")
    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():
    """主函数"""
    try:
        print("开始测试 OceanBase 数据库 Oracle 模式连接...")
        
        # 测试各种数据类型
        test_char_types()
        test_number_types()
        test_time_types()
        test_lob_types()
        
        print("\n所有测试完成！")
        
    except Exception as e:
        print(f"测试过程中发生错误: {e}")
    finally:
        # 关闭数据库连接
        if conn:
            conn.close()
            print("数据库连接已关闭")

if __name__ == "__main__":
    main()

python3 test_oracle.py

import cx_Oracle
from cx_Oracle import SessionPool

def test_simple_pool():
    """连接池测试"""
    try:
        print("创建连接池...")
        # 创建连接池
        pool = SessionPool(
            user="test@oracle",
            password="test",
            dsn="xxx.xxx.xxx.xxx:2881/TEST",
            min=2,
            max=5,
            increment=1
        )
        print(f"连接池创建成功！")
        print(f"- 最小连接数: {pool.min}")
        print(f"- 最大连接数: {pool.max}")
        print(f"- 增量: {pool.increment}")
        
        print("\n测试连接池使用...")
        # 从连接池获取连接
        with pool.acquire() as connection:
            with connection.cursor() as cursor:
                # 执行查询
                cursor.execute("SELECT COUNT(*) FROM test_char")
                result = cursor.fetchone()
                print(f"test_char 表记录数: {result[0]}")
                
                # 执行另一个查询
                cursor.execute("SELECT * FROM test_char WHERE id = 1")
                data = cursor.fetchone()
                print(f"查询结果: {data}")
        
        print(f"\n连接池状态:")
        print(f"- 当前打开的连接数: {pool.opened}")
        print(f"- 当前繁忙的连接数: {pool.busy}")
        
        print("\n连接池测试成功完成！")
        # 注意：不调用 pool.close() 来避免段错误
        
    except Exception as e:
        print(f"发生错误: {e}")

if __name__ == "__main__":
    test_simple_pool()