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

Ubuntu系统下Python连接国产KingbaseES数据库实现增删改查

backend 2025/9/8 5:28:46

摘要：本文将介绍Ubuntu系统下如何使用Python连接国产金仓数据库KingbaseES，并实现基本的增删改查操作。文中将通过具体代码示例展示连接数据库、执行SQL语句以及处理结果的全过程。这里把Python连接KingbaseES的经验整理一下，希望能帮到同样踩坑的兄弟。

1.环境准备与驱动安装

1.1 科普ksycopg2知识

1.2 官方下载ksycopg2驱动

1.3 安装ksycopg2驱动

2. 连接KingbaseES数据库

3. 创建数据表

4. 实现增删改查功能

4.1 新增

4.2 查询

4.3 修改

4.4 删除

4.5 封装一个类crud方便复用

5.总结

1.环境准备与驱动安装

KingbaseES提供了专门的Python驱动包ksycopg2，它是基于Python DB API 2.0规范实现的线程安全数据库适配器！

1.1 科普ksycopg2知识

ksycopg2是Python编程语言的KingbaseES数据库适配器。它的主要特点是Python DB API 2.0 规范的完整实现和线程安全。

ksycopg2 主要在C程序中作为libkci包装器实现，因此既高效又安全。它拥有客户端和服务端游标，支持异步通信和通知、复制。

ksycopg2驱动需要和python大版本一致，如python3.8的ksycopg2驱动支持python3.8.x的任意小版本。

更多科普请查看金仓KingbaseES官方手册：2. 概述 — KingbaseES产品手册

1.2 官方下载ksycopg2驱动

1、首先需要下载并安装与你的Python版本和系统架构匹配的ksycopg2驱动。驱动可以从KingbaseES官方网站（KES-电科金仓官网）获取，如下图所示：

2、这里兄弟盟根据自己电脑系统，选择对应的版本，我这里是linux，下载下来如下图所示：KingbaseES_V009R001C010B0004_X64_Linux_Ksycopg2.tar 解压后可以看到有python2.7、python3.6、python3.7、python3.8、python3.9、python3.10、python3.11、python3.12，准备得真是周到，照顾各位大佬电脑上不同python版本，这一点为国产金仓数据库点赞👍👍👍

1.3 安装ksycopg2驱动

1、上面下载后解压并将ksycopg2文件夹放置在Python的模块搜索路径中，如果不清楚自己Python的模块在哪里，可以写个简单python代码查看：

import sys
print(sys.path)

2、运行后如下所示，可以查看Python的模块位置：/usr/lib/python3/dist-packages

3、开始上传到：/usr/lib/python3/dist-packages

4、此外，还需要将KingbaseES的libkci库文件路径添加到LD_LIBRARY_PATH环境变量中：

export LD_LIBRARY_PATH=/kingbase/data/KESRealPro/V009R002C012/Server/lib:$LD_LIBRARY_PATH

如果不清楚自己KingbaseES的libkci库文件路在哪里，可以用这个命令查看：

ps -ef | grep kingbase

5、验证安装ksycoph2驱动

import ksycopg2
print("ksycopg2驱动安装成功")

2. 连接KingbaseES数据库

使用ksycopg2连接KingbaseES数据库需要提供数据库名称、用户名、密码、主机地址和端口号等信息。

import ksycopg2def create_connection():try:conn = ksycopg2.connect(database="TEST",user="SYSTEM",password="qwe123!@#",host="127.0.0.1",port="54321")print("数据库连接成功")return connexcept Exception as e:print(f"连接数据库失败: {e}")return None# 建立数据库连接
connection = create_connection()

创建一个connect_database.py把上面代码复制进去，然后执行

python connect_database.py

3. 创建数据表

在执行增删改查操作前，需要先创建一张测试表，有表才好对后面的案例进行操作。

def create_table(conn):try:cursor = conn.cursor()create_table_sql = """CREATE TABLE IF NOT EXISTS user_info (id INTEGER PRIMARY KEY,username VARCHAR(50) NOT NULL,age INTEGER,created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP)"""cursor.execute(create_table_sql)conn.commit()cursor.close()print("表创建成功")except Exception as e:print(f"创建表失败: {e}")conn.rollback()# 创建表
if connection:create_table(connection)

同理，在ubuntu服务器上新建一个create_table.py文件，把上面代码丢进去执行：

python create_table.py

4. 实现增删改查功能

4.1 新增

def insert_data_example():"""插入数据示例"""db_manager = KingbaseESManager(dbname="test",user="SYSTEM",password="qwe123!@#",host="127.0.0.1",port="54321")if db_manager.connect():# 插入单条数据insert_sql = "INSERT INTO user_info (id, username, age) VALUES (%s, %s, %s)"params = (1, '张三', 25)db_manager.execute_update(insert_sql, params)# 插入多条数据users_to_insert = [(2, '李四', 30),(3, '王五', 28),(4, '赵六', 35),(5, '钱七', 22)]for user in users_to_insert:db_manager.execute_update(insert_sql, user)db_manager.disconnect()# 执行插入数据
insert_data_example()

调用python后登录数据库查看是否已经插入数据库，ubuntu登录如下所示

# 切换到kingbase用户
su - kingbase# 连接数据库
ksql -U SYSTEM -d test -p 54321# 如果上面执行不成功，可以指定目录来连接
/kingbase/data/KESRealPro/V009R002C012/Server/bin/ksql -U SYSTEM -d test -p 54321

执行查询语句，查询所有用户数据，验证插入结果，如下图所示：

-- 查询所有用户数据，验证插入结果
SELECT * FROM user_info ORDER BY id;

4.2 查询

def query_data_example():"""查询数据示例"""db_manager = KingbaseESManager(dbname="test",user="SYSTEM",password="qwe123!@#",host="127.0.0.1",port="54321")if db_manager.connect():# 查询所有数据print("所有用户信息:")select_all_sql = "SELECT * FROM user_info ORDER BY id"all_users = db_manager.execute_query(select_all_sql)for user in all_users:print(user)# 根据ID查询特定用户print("\n查询ID为2的用户:")select_by_id_sql = "SELECT * FROM user_info WHERE id = %s"user_by_id = db_manager.execute_query(select_by_id_sql, (2,))print(user_by_id)# 根据年龄范围查询用户print("\n年龄在25-30岁之间的用户:")select_by_age_sql = "SELECT * FROM user_info WHERE age BETWEEN %s AND %s ORDER BY age"users_by_age = db_manager.execute_query(select_by_age_sql, (25, 30))for user in users_by_age:print(user)db_manager.disconnect()# 执行查询数据
query_data_example()

登录数据库验证如下所示：

4.3 修改

def update_data_example():"""更新数据示例"""db_manager = KingbaseESManager(dbname="TEST",user="SYSTEM",password="your_password",host="127.0.0.1",port="54321")if db_manager.connect():# 更新用户年龄print("更新张三的年龄为26:")update_sql = "UPDATE user_info SET age = %s WHERE id = %s"db_manager.execute_update(update_sql, (26, 1))# 验证更新结果print("\n更新后的用户信息:")select_sql = "SELECT * FROM user_info WHERE id = %s"updated_user = db_manager.execute_query(select_sql, (1,))print(updated_user)db_manager.disconnect()# 执行更新数据
update_data_example()

执行python脚本后，登录数据库后台查看，确实是每个用户小于30岁的都加了1岁，如下所示：

4.4 删除

def delete_data_example():"""删除数据示例"""db_manager = KingbaseESManager(dbname="TEST",user="SYSTEM",password="your_password",host="127.0.0.1",port="54321")if db_manager.connect():# 删除ID为5的用户print("删除ID为5的用户:")delete_sql = "DELETE FROM user_info WHERE id = %s"db_manager.execute_update(delete_sql, (5,))# 验证删除结果print("\n删除后的所有用户:")select_all_sql = "SELECT * FROM user_info ORDER BY id"remaining_users = db_manager.execute_query(select_all_sql)for user in remaining_users:print(user)db_manager.disconnect()# 执行删除数据
delete_data_example()

同理，验证数据如下所示：

4.5 封装一个类crud方便复用

下面将实现完整的增删改查功能，并将这些操作封装在一个类中，方便复用。

import ksycopg2
from datetime import datetimeclass KingbaseESManager:def __init__(self, dbname, user, password, host, port):self.conn = Noneself.db_params = {"database": TEST,"user": SYSTEM,"password": qwe123!@#,"host": 127.0.0.1,"port": 54321}def connect(self):"""连接数据库"""try:self.conn = ksycopg2.connect(**self.db_params)print("数据库连接成功")return Trueexcept Exception as e:print(f"连接数据库失败: {e}")return Falsedef disconnect(self):"""断开数据库连接"""if self.conn:self.conn.close()print("数据库连接已关闭")def execute_query(self, sql, params=None):"""执行查询语句并返回结果"""try:cursor = self.conn.cursor()if params:cursor.execute(sql, params)else:cursor.execute(sql)results = cursor.fetchall()cursor.close()return resultsexcept Exception as e:print(f"查询执行失败: {e}")return Nonedef execute_update(self, sql, params=None):"""执行更新操作（插入、更新、删除）"""try:cursor = self.conn.cursor()if params:cursor.execute(sql, params)else:cursor.execute(sql)self.conn.commit()affected_rows = cursor.rowcountcursor.close()print(f"操作成功，影响行数: {affected_rows}")return affected_rowsexcept Exception as e:self.conn.rollback()print(f"操作执行失败: {e}")return -1def insert_user(self, id, username, age):"""插入用户数据"""sql = "INSERT INTO user_info (id, username, age) VALUES (%s, %s, %s)"params = (id, username, age)return self.execute_update(sql, params)def select_all_users(self):"""查询所有用户"""sql = "SELECT * FROM user_info ORDER BY id"return self.execute_query(sql)def select_user_by_id(self, id):"""根据ID查询用户"""sql = "SELECT * FROM user_info WHERE id = %s"params = (id,)return self.execute_query(sql, params)def update_user_age(self, id, new_age):"""更新用户年龄"""sql = "UPDATE user_info SET age = %s WHERE id = %s"params = (new_age, id)return self.execute_update(sql, params)def delete_user(self, id):"""删除用户"""sql = "DELETE FROM user_info WHERE id = %s"params = (id,)return self.execute_update(sql, params)def search_users_by_age_range(self, min_age, max_age):"""根据年龄范围查询用户"""sql = "SELECT * FROM user_info WHERE age BETWEEN %s AND %s ORDER BY age"params = (min_age, max_age)return self.execute_query(sql, params)# 使用示例
if __name__ == "__main__":# 创建数据库管理实例db_manager = KingbaseESManager(dbname="TEST",user="SYSTEM",password="qwe123!@#",host="127.0.0.1",port="54321")# 连接数据库if db_manager.connect():# 插入多条用户数据users_to_insert = [(1, '张三', 25),(2, '李四', 30),(3, '王五', 28),(4, '赵六', 35),(5, '钱七', 22)]for user in users_to_insert:db_manager.insert_user(*user)# 查询所有用户print("所有用户信息:")all_users = db_manager.select_all_users()for user in all_users:print(user)# 根据ID查询用户print("\n查询ID为2的用户:")user_by_id = db_manager.select_user_by_id(2)print(user_by_id)# 更新用户年龄print("\n更新张三的年龄为26:")db_manager.update_user_age(1, 26)# 查询年龄在25-30岁之间的用户print("\n年龄在25-30岁之间的用户:")users_by_age = db_manager.search_users_by_age_range(25, 30)for user in users_by_age:print(user)# 删除ID为5的用户print("\n删除ID为5的用户:")db_manager.delete_user(5)# 再次查询所有用户print("\n删除后的所有用户:")remaining_users = db_manager.select_all_users()for user in remaining_users:print(user)# 断开连接db_manager.disconnect()

5.总结

本文介绍了在Ubuntu系统中使用Python连接国产金仓数据库KingbaseES的方法。主要内容包括：1.安装与Python版本匹配的ksycopg2驱动；2.配置环境变量和连接参数；3.实现数据库连接、建表及增删改查操作；4.封装可复用的数据库操作类。通过具体代码示例演示了数据插入、查询、更新和删除等常见操作，并提供了验证方法。文章还分享了环境配置中的常见问题和解决方案，帮助开发者快速上手KingbaseES数据库操作。

编码不易，希望各位点赞支持，也支持我们国产的数据库，纸上得来终觉浅，希望各位大佬也亲自动手尝试一下，如果遇到什么问题欢迎评论区留言交流，相互学习，共同进步~正在走向自律