在企业级数据库开发场景中，OceanBase Oracle 兼容模式凭借稳定的性能、完美的Oracle语法适配能力，被广泛应用于金融、政务、互联网核心业务系统。
而Python作为数据分析、后端开发、自动化运维的主流语言，实现其与OceanBase Oracle模式的稳定连接，是日常开发、数据同步、业务测试的必备技能。
今天给大家带来一套CentOS 7 环境从零搭建的标准实操教程！
实验前置
Q1：关于python环境的选择
A1：根据文档建议是Python 3.7 或更高版本，根据最新的obci版本要求，建议是3.9，或者根据本文选择主流的3.10版本即可
Q2：各文件获取方式和介绍
| | |
|---|
| oracle-instantclient12.2-basic-12.2.0.1.0-1.x86_64.rpm | Oracle OCI头文件，obci需要依赖里面的框架 | https://www.oracle.com/database/technologies/instant-client/linux-x86-64-downloads.html#tab-id-5de3f833-4511-46de-aa28-a8179e6f6ed7 |
| libobclient-2.2.13-1.el7.x86_64.rpm | | https://www.oceanbase.com/softwarecenter或者https://github.com/oceanbase/obconnector-c/releases |
| obci-2.1.1.3-22026051517.el7.x86_64.rpm | | 目前只有老版本的obci释出，并没有适配到python环境https://www.oceanbase.com/docs/common-oceanbase-cloud-1000000000095858 需通过工单或者架构师获取 |
Q3：除了本文的Linux环境下python连接oboracle，Windows下的主流环境是否支持？
A3：有相关的驱动支持的，具体可联系售后或方案架构师。
实验步骤
1、测试环境：Centos7
[root@iZwz9ak4ce2et9ihb25omrZ ~]# cat /etc/system-release
CentOS Linux release 7.9.2009 (Core)
2、安装python新版本
建议安装前查看系统原有的python相关
python --version
python2 --version
python3 --version
[root@iZwz9ak4ce2et9ihb25omrZ ~]# python --version
Python 2.7.5
[root@iZwz9ak4ce2et9ihb25omrZ ~]# python2 --version
Python 2.7.5
[root@iZwz9ak4ce2et9ihb25omrZ ~]# python3 --version
Python 3.6.8
使用二进制编译安装新一点的python版本
官网二进制下载地址：https://www.python.org/ftp/python/3.10.20/Python-3.10.20.tgz
# 安装目录
./configure --enable-optimizations --prefix=/usr/local/python3.10
make -j$(nproc)
make altinstall
cat >> /etc/profile <
export PATH=/usr/local/python3.10/bin:$PATH
EOF
source /etc/profile
sudo ln -s /usr/local/python3.10/bin/python3.10 /usr/local/bin/python3
sudo ln -s /usr/local/python3.10/bin/pip3.10 /usr/local/bin/pip3
[root@iZwz9ak4ce2et9ihb25omrZ ~]# pip3 --version
pip 23.0.1 from /usr/local/python3.10/lib/python3.10/site-packages/pip (python 3.10)
[root@iZwz9ak4ce2et9ihb25omrZ ~]# python3 --version
Python 3.10.20
2、准备对应的rpm包进行安装
rpm -ivh
3、设置前面安装好的C动态库的环境变量
cat >> /etc/profile <
export LD_LIBRARY_PATH=/u01/obclient/lib/:/usr/lib/oracle/12.2/client64/lib/:$LD_LIBRARY_PATH
EOF
source /etc/profile
4、开启一个venv的py项目开始测试
python3 -m venv project1
cd project1
source bin/activate
查看阅读驱动附加的readme信息
(project1) [root@iZwz9ak4ce2et9ihb25omrZ project1]# cat /u01/obclient/python/
cx_Oceanbase-6.1.0.tar.gz cx_Oracle-6.1.0.tar.gz cx_Oracle-8.3.0.tar.gz readme.txt
cx_Oceanbase-8.3.0.tar.gz cx_Oracle-7.3.0.tar.gz oracledb-3.3.0.tar.gz
(project1) [root@iZwz9ak4ce2et9ihb25omrZ project1]# cat /u01/obclient/python/readme.txt
obci安装后/u01/obclient/python下面有两个包，cx_Oracle/cx_Oceanbase(代码相同都是用于链接ob oracle)
1. 如果应用只链接ob oracle,建议用/u01/obclient/python/cx_Oracle。
2. 如果应用同时链接oracle/oboracle，建议用/u01/obclient/python/cx_Oceanbase，然后oracle用原生的cx_Oracle。
oracledb3.3.0编译（python3.9以上）
pip install cython
pip install setuptools wheel
linux:
export CFLAGS="-std=c99"
windows:
set CFLAGS=-std=c99
加载驱动包，这里选择cx_Oracle
(project1) [root@iZwz9ak4ce2et9ihb25omrZ project1]# tar -zxvf /u01/obclient/python/cx_Oracle-8.3.0.tar.gz -C .
(project1) [root@iZwz9ak4ce2et9ihb25omrZ project1]# ls -l cx_Oracle-8.3.0/
total 60
-rwxrwxrwx 1 root root 284 Feb 10 2025 CONTRIBUTING.md
drwxrwxrwx 3 root root 4096 Mar 16 13:34 doc
-rwxrwxrwx 1 root root 1841 Apr 15 2024 LICENSE.txt
-rwxrwxrwx 1 root root 256 Apr 15 2024 MANIFEST.in
drwxrwxrwx 10 root root 4096 Mar 16 13:34 odpi
-rwxrwxrwx 1 root root 103 Apr 15 2024 pyproject.toml
-rwxrwxrwx 1 root root 2862 Feb 10 2025 README.md
-rwxrwxrwx 1 root root 285 Apr 15 2024 README.txt
drwxrwxrwx 4 root root 4096 Mar 16 13:34 samples
-rwxrwxrwx 1 root root 1738 Feb 10 2025 SECURITY.md
-rwxrwxrwx 1 root root 1121 Apr 15 2024 setup.cfg
-rwxrwxrwx 1 root root 2581 Apr 9 15:47 setup.py
drwxrwxrwx 2 root root 4096 Mar 16 13:34 src
drwxrwxrwx 4 root root 4096 Mar 16 13:34 test
-rwxrwxrwx 1 root root 387 Apr 15 2024 tox.ini
# 进入解压后的目录
cd cx_Oracle-8.3.0
# 安装 cx_Oracle 驱动
python3 setup.py install
# 已安装
(project1) [root@iZwz9ak4ce2et9ihb25omrZ project1]# pip list |grep cx
cx_Oracle 8.3.0.post84
测试连接demo，注意配置好环境和数据库的连通性
import cx_Oracle
# 数据库连接信息
username = 'user'
password = 'pass'
oracle_connection = 'host:port/DB_NAME'
# 创建数据库连接
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()
(project1) [root@iZwz9ak4ce2et9ihb25omrZ project1]# python3 test_oracle.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')
查询结果: [(, , b'1234')]
所有测试完成！
数据库连接已关闭
参考文档
Python 驱动连接 OceanBase 数据库 Oracle 模式指南
https://www.oceanbase.com/docs/common-oceanbase-database-cn-1000000004802405
安装 OBCI
https://www.oceanbase.com/docs/enterprise-obci-cn-1000000003572580
