--- title: "Python 驱动连接 OceanBase 数据库 Oracle 模式指南 - OceanBase 数据库 V4.6.0 | OceanBase 文档中心" description: Python 驱动连接 OceanBase 数据库 Oracle 模式指南 本文档介绍如何通过 Python cx_Oracle 驱动连接和使用 OceanBase 数据库 Oracle 模式。cx_Oracle 是 Oracle 官方提供的 Python 数据库接口，与 Oracle OCI 兼容，可以无缝连接 O… --- 切换语言 - 中文站 - 简体中文 - International - English - 日本站 - 日本語 文档反馈![](https://mdn.alipayobjects.com/huamei_22khvb/afts/img/A*P8CuR4UJ_FkAAAAAAAAAAAAADiGDAQ/original) OceanBase 数据库分布式版 - V 4.6.0 # Python 驱动连接 OceanBase 数据库 Oracle 模式指南 更新时间：2026-07-15 15:27:38 [编辑](https://github.com/oceanbase/oceanbase-doc/edit/V4.6.0/zh-CN/300.develop/200.application-development-of-oracle-mode/200.sample-program-of-oracle-mode/200.python-of-oracle-mode/100.obci-python.md) 本文档介绍如何通过 Python cx_Oracle 驱动连接和使用 OceanBase 数据库 Oracle 模式。cx_Oracle 是 Oracle 官方提供的 Python 数据库接口，与 Oracle OCI 兼容，可以无缝连接 OceanBase 数据库的 Oracle 租户。 ## 前提条件 - 您已安装 Python 3.7 或更高版本。 - 您已安装 OceanBase 数据库并且创建了 Oracle 模式租户。 - 您已安装 libobclient 和 OBCI 组件。 ## 操作步骤 按以下步骤操作，使用 cx_Oracle 驱动连接 OceanBase 数据库 Oracle 模式。 ### 步骤一：安装 Python 3.7 环境 确保系统已安装 Python 3.7 或更高版本。您可以通过以下命令检查 Python 版本： ```shell python3 --version ``` 如果未安装，请根据您的操作系统安装 Python 3.7。 ### 步骤二：安装 libobclient 和 OBCI 组件 安装 OceanBase 客户端组件，包括 libobclient 和 OBCI： ```shell # 卸载旧版本（如果存在） 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 ``` #### 注意 请确保 libobclient 版本 >= 2.2.11，obci 版本 >= 2.1.1。版本过低可能导致连接失败或功能异常。 ### 步骤三：安装 cx_Oracle 驱动 解压并安装 cx_Oracle 驱动： ```shell # 切换到驱动安装目录 cd /directory/of/cx_Oracle-8.3.0.tar.gz # 解压 cx_Oracle 驱动包 tar -xvf cx_Oracle-8.3.0.tar.gz # 进入解压后的目录 cd cx_Oracle-8.3.0 # 安装 cx_Oracle 驱动 python3 setup.py install ``` 安装完成后，系统会在 `/usr/local/lib/python3.7/site-packages/` 目录下生成 cx_Oracle 模块。 ### 步骤四：配置环境变量 设置 LD_LIBRARY_PATH 环境变量，确保 Python 能够找到 OceanBase 客户端库： ```shell export LD_LIBRARY_PATH=/u01/obclient/lib/:$LD_LIBRARY_PATH ``` 建议将此配置添加到您的 shell 配置文件中（如 `~/.bashrc` 或 `~/.zshrc`），以便永久生效。 ### 步骤五：编写测试程序 创建一个名为 `test_oracle.py` 的文件，包含以下内容： ```python 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() ``` ### 步骤六：运行测试程序 在命令行中运行测试程序： ```shell python3 test_oracle.py ``` 程序将依次测试字符类型、数值类型、时间类型和 LOB 类型的数据操作，并输出执行结果。 ## 连接池支持 从 OBCI 2.1.1 版本开始，支持连接池功能。您可以使用连接池来提高应用程序的性能和并发处理能力。 ```python 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() ``` 上一篇 下一篇 ![有帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*y6ocSqN8cqsAAAAAAAAAAAAAARQnAQ)![无帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*BG9IQJyLHF8AAAAAAAAAAAAAARQnAQ)![反馈](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*eTWdQKCRKHwAAAAAAAAAAAAAARQnAQ)[AI](https://www.oceanbase.com/obi) 咨询热线