Python 应用连接 OceanBase 数据库 Oracle 模式租户使用指南

主要介绍 Python 应用如何连接到 OceanBase 数据库 Oracle 模式租户。

操作指南

1. 前提条件。

   • 已安装 Python 3.7 或更高版本。

   • 已安装 OceanBase 数据库并且创建了 Oracle 模式租户。

   • 已安装 libobclient 和 OBCI 组件（版本需要满足前述的适用版本条件）。

2. 安装 cx_Oracle 驱动。

   进入驱动安装目录（默认 /u01/obclient/python/ 目录）：

   [root@ob51 ~]# cd /u01/obclient/python
   [root@ob51 python]# ll
   total 3588
   -rwxr-xr-x 1 root root  954434 Jul 10 11:01 cx_Oracle-6.1.0.tar.gz
   -rwxr-xr-x 1 root root 1206369 Jul 10 11:01 cx_Oracle-7.3.0.tar.gz
   -rwxr-xr-x 1 root root 1506405 Jul 10 11:01 cx_Oracle-8.3.0.tar.gz
   
   ### 解压 cx_Oracle 驱动包
   [root@ob51 python]# tar -xvf cx_Oracle-8.3.0.tar.gz
   
   ### 进入解压后的目录
   [root@ob51 python]# cd cx_Oracle-8.3.0/
   
   ### 安装 cx_Oracle 驱动 
   python3 setup.py install
   
   ### 强烈建议使用自带的 cx_Oracle 驱动，使用三方下载的驱动可能会导致安装失败
   

3. 配置环境变量。

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

4. 编写测试程序。

   import cx_Oracle
   
   # 数据库连接信息
   username = 'TEST_ZQ@test_zq'
   password = 'xxx'
   oracle_connection = 'xx.xxx.211.54:2881/test_zq'
   
   # 创建数据库连接
   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()
   

5. 运行测试。

   [root@ob51 ~]# python3 test.py
   开始测试 OceanBase 数据库 Oracle 模式连接...
   
   === 测试字符类型 ===
   SQL 执行成功: DROP TABLE test_char
   SQL 执行成功:
        CREATE TABLE test_char (
            id INT,
            a VARCHAR2(20),
            b CHAR(10),
            c NCHAR(10),
            d NVARCHAR2(10)
        )
   
   SQL 执行成功: INSERT INTO test_char VALUES (1, 'hello', 'adffdf', '2df4d', 'dsf44f')
   查询结果: [(1, 'hello', 'adffdf    ', '2df4d     ', 'dsf44f')]
   
   === 测试数值类型 ===
   SQL 执行成功: DROP TABLE test_number
   SQL 执行成功:
        CREATE TABLE test_number (
            a NUMBER,
            b FLOAT(126),
            c BINARY_FLOAT,
            d BINARY_DOUBLE
        )
   
   SQL 执行成功: INSERT INTO test_number VALUES (12.32, 12.34, 14.23, 123.3433)
   查询结果: [(12.32, 12.34, 14.229999542236328, 123.3433)]
   
   === 测试时间类型 ===
   SQL 执行成功: DROP TABLE test_time
   SQL 执行成功:
        CREATE TABLE test_time (
            a DATE,
            b TIMESTAMP,
            c TIMESTAMP WITH TIME ZONE,
            d TIMESTAMP WITH LOCAL TIME ZONE
        )
   
   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'
        )
   
   查询结果: [(datetime.datetime(2022, 8, 29, 14, 44, 30), datetime.datetime(2022, 8, 29, 14, 44, 30), datetime.datetime(2022, 8, 29, 14, 44, 30), datetime.datetime(2022, 8, 29, 14, 44, 30))]
   
   === 测试 LOB 类型 ===
   SQL 执行成功: DROP TABLE test_lob
   SQL 执行成功:
        CREATE TABLE test_lob (
            a CLOB,
            b BLOB,
            c RAW(100)
        )
   
   SQL 执行成功: INSERT INTO test_lob VALUES ('sdfdslkfjldsf', '31323334353637', '31323334')
   查询结果: [(<cx_Oracle.LOB object at 0x7fdfaa008ba0>, <cx_Oracle.LOB object at 0x7fdfaa008c00>, b'1234')]
   
   所有测试完成，数据库连接已关闭。
   

适用版本

• OceanBase 数据库 V2.x、V3.x、V4.x 版本。

• libobclient V2.2.11 及之后版本。

• obci V2.1.1 BP1 及之后版本。

参考链接

详情参见：Python 驱动连接 OceanBase 数据库 Oracle 模式指南。