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

Ne0inhk

14 Mar 2026 — 12 min read

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

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 ksycopg2 def create_connection(): try: conn = ksycopg2.connect( database="TEST", user="SYSTEM", password="qwe123!@#", host="127.0.0.1", port="54321" ) print("数据库连接成功") return conn except 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 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 datetime class KingbaseESManager: def __init__(self, dbname, user, password, host, port): self.conn = None self.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 True except Exception as e: print(f"连接数据库失败: {e}") return False def 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 results except Exception as e: print(f"查询执行失败: {e}") return None def 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.rowcount cursor.close() print(f"操作成功，影响行数: {affected_rows}") return affected_rows except Exception as e: self.conn.rollback() print(f"操作执行失败: {e}") return -1 def 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数据库操作。

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