当前位置：首页 > news >正文

python连接Oceanbase

news 2026/7/7 15:45:58

python连接Oceanbase

一、OB简介

OceanBase（OB）是蚂蚁集团完全自主研发、无开源依赖的国产原生分布式关系型数据库。现在官方最新版本是4.6.0版本，主推的比较稳定的版本是4.4版本。4.2版本之前主要考量的使用场景是OLTP的实时响应。随着数据仓库、数据中台等的使用，4.3版本后逐步开始OLAP的业务诉求（支持列存表），支持高可用和分布式，可横向拓展，4.6版本还支持向量存储。

OB支持多租户，租户类似于传统数据库的数据库实例，租户通过资源池与资源关联，从而独占一定的资源配额，可以动态调整资源配额。在租户下可以创建 Database、表、用户等数据库对象。（可以简单理解为oracle的C版本中的CDB和PDB，也就是多实例，只不过为了迎合国产化的进程在租户上做了适配性的设置，做快的理解就是支持多实例）。OB支持两种租户模式MYSQL模式和ORACLE模式。

接下来的内容以oracle租户为例（因为目前信创更多的场景是寻求oracle的替代，这样切换会比较平滑，技术上手也会更快）。

查看当前数据库版本号

SHOWVARIABLESLIKE'version';

查看租户模式

SHOWVARIABLESLIKE'ob_compatibility_mode';

如果数据库是4.2版本或者更早的版本建议升级到4.4以上的版本。

二、连接数据库

2.1 利用oracledb连接数据库

首先宣布结果：连接失败。

选用这个第三方包是因为它与cx_Oracle相比支持thin模式，不需要依赖于Instant Client，因为我的服务器不在自己手里，服务器相关操作需要依赖于跨部门的同事，所以选择用这个包。

importoracledb# OB连接信息username="用户名@租户名"password="用户密码"host="127.0.0.1"port=2883service_name="服务名"defconnect_ob():conn=Nonetry:# Thin模式默认启用，无需初始化客户端conn=oracledb.connect(user=username,password=password,host=host,port=port,service_name=service_name)print("数据库连接成功")print(f"数据库版本:{conn.version}")# 简单查询测试cur=conn.cursor()cur.execute("SELECT SYSDATE FROM DUAL")res=cur.fetchone()print("当前时间:",res[0])exceptoracledb.Errorase:error,=e.argsprint(f"连接失败:{error.code},{error.message}")finally:ifconn:conn.close()print("连接已关闭")if__name__=="__main__":test_ob_thin()

2.2 利用cx_oracle连接

经历上面的步骤决定按照官方文档一步步来了，用厚切模式连接。然而…

官方文档写明可以连接但是需要使用它自带的cx_oracle，因为底层依赖的类库不一样，oracle连接依赖的是oci，而oceanbase依赖的是obci，官方文档做了如下说明

### 强烈建议使用自带的 cx_Oracle 驱动，使用三方下载的驱动可能会导致安装失败

如果本机已有环境已经在运行cx_oracle连接oracle的程序，不建议卸载已有包，安装OB自带，因为底层不一样。

PS：都已经知道底层不一样了知道做一个第三方，为什么非要杂糅到已有的cx_oracle，为何不单独设第三方包这是我始终未能理解的。cx_ob_oracle不是也很好吗。

建议同一台服务器安装两个版本的python这样不会影响到已有程序运行。

环境检查，仅支持python3.7以上的版本
```
python3--version
```
安装 libobclient 和 OBCI 组件
libobclient是客户端包，社区也可以支持下载，但是OBCI必须是企业账号才可以下载。
libobclient 版本 >= 2.2.11，obci 版本 >= 2.1.1。版本过低可能导致连接失败或功能异常
卸载旧版本（如果存在）
```
rpm-elibobclientrpm-eobci4.rpm
```
安装新版本
```
rpm-ivhlibobclient-2.2.11-42025062010.el7.x86_64.rpmrpm-ivhobci-2.1.1-342025070917.el7.x86_6
```
安装cx_Oracle
以上步骤安装完成后，会在python的安装目录下找到自动生成的 cx_Oracle 模块，3.7版本在/usr/local/lib/python3.7/site-packages/目录下。
解压
```
tar-xvfcx_Oracle-8.3.0.tar.gz
```
进入安装目录
```
cdcx_Oracle-8.3.0
```
安装
```
python3 setup.pyinstall
```
配置环境变量
确保可以找到OB客户端
打开 ~/.bashrc 文件将环境变量写入
```
exportLD_LIBRARY_PATH=/user/local/obclient/lib/:$LD_LIBRARY_PATH
```

代码案例

importcx_Oracle# Linux 配置动态库路径 LD_LIBRARY_PATH（如果配置了环境变量也可以去掉这部分）lib_path="/usr/local/obclient/lib"os.environ["LD_LIBRARY_PATH"]=lib_path 数据库连接信息 username='用户名@租户名'password='用户密码'oracle_connection='127.0.0.1:2883/服务名'创建数据库连接 conn=cx_Oracle.connect(username,password,oracle_connection)defconnect_ob(sql):"""查询并打印数据"""cur=conn.cursor()try:cur.execute(sql)data=cur.fetchall()print(f"查询结果:{data}")returndataexceptExceptionase:print(f"查询失败:{sql}")print(f"错误信息:{e}")returnNonefinally:cur.close()defmain():"""主函数"""try:print("开始测试 OceanBase 数据库 Oracle 模式连接...")connect_ob("SELECT SYSDATE FROM DUAL")print("测试完成！")exceptExceptionase:print(f"测试过程中发生错误:{e}")finally:ifconn:conn.close()print("数据库连接已关闭")if__name__=="__main__":main()

2.3 用jdbc连接

因为环境的特点，我目前没有单独的环境可以操作，只能合用其他环境，为了不影响已有的oracle连接脚本运行（也因为我自己操作不了服务器），所以找到了这种方式。

jdbc连接不需要依赖于其他，只要有python运行环境（jdk 1.8以上），然后有对应的jar包就可以。这里又不理解了jar包也是企业用户才可以下载…

这里使用maven下载对应的版本（我这里数据库的版本为4.2版本）

iwr-Uri"https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.15/oceanbase-client-2.4.15.jar"-OutFile"oceanbase-client-2.4.15.jar"

然后下载第三方包

pipinstalljaydebeapiJPype1==1.5.0

代码样例

importjaydebeapi JDBC_JAR_PATH="/opt/ob_jdbc/oceanbase-client-2.4.15.jar"OB_HOST="127.0.0.1"OB_PORT=2883OB_USER="用户名@租户名#集群名"OB_PWD="密码"OB_DRIVER_CLASS="com.alipay.oceanbase.jdbc.Driver"# 拼接JDBC连接串jdbc_url=f"jdbc:oceanbase://{OB_HOST}:{OB_PORT}?useUnicode=true&characterEncoding=UTF-8"try:# 创建连接conn=jaydebeapi.connect(jclassname=OB_DRIVER_CLASS,url=jdbc_url,driver_args=[OB_USER,OB_PWD],jars=JDBC_JAR_PATH)print("OceanBase Oracle租户 连接成功！")# 测试查询Oracle标准SQLcur=conn.cursor()cur.execute("SELECT SYSDATE FROM DUAL")print("DUAL查询结果：",cur.fetchone())# 批量插入示例batch_sql="INSERT INTO TEST.T1(ID, NAME) VALUES (?, ?)"batch_data=[(1001,"测试1"),(1002,"测试2")]cur.executemany(batch_sql,batch_data)conn.commit()print("批量插入完成")# 关闭资源cur.close()conn.close()exceptExceptionase:print("连接失败，错误信息：",str(e))