在国产化替代浪潮下，国产数据库KingbaseES（人大金仓）凭借其自主可控特性，在政务、金融等领域广泛应用。本文以Ubuntu系统为环境，通过Python实现与KingbaseES的连接，并完成完整的增删改查（CRUD）操作，为开发者提供可直接复用的技术方案。
一、环境准备与驱动安装
KingbaseES基于PostgreSQL协议开发，推荐使用官方提供的
ksycopg2驱动（兼容Python DB API 2.0规范）。以Ubuntu 20.04系统为例，具体步骤如下：
-
下载驱动包
从金仓官网下载与Python版本匹配的驱动包（如KingbaseES_V009R001C010B0004_X64_Linux_Ksycopg2.tar<"www.gov.cn.huizhou.manct.cn">），解压后包含多版本Python适配目录。 -
配置模块路径
将对应Python版本的ksycopg2文件夹复制至/usr/lib/python3/dist-packages/（通过import sys; print(sys.path)可查看模块搜索路径）。 -
设置动态库路径
KingbaseES依赖libkci库文件，需将其路径添加至LD_LIBRARY_PATH环境变量：bash1export LD_LIBRARY_PATH=<"www.gov.cn.zhuhai.manct.cn">/kingbase/data/KESRealPro/V009R002C012/Server/lib:$LD_LIBRARY_PATH
-
验证安装
启动Python交互环境，执行import ksycopg2，无报错即表示安装成功。
二、数据库连接与表结构初始化
1. 建立连接
通过
ksycopg2.connect()方法传入连接参数，包含数据库名、用户名、密码、主机地址及端口（默认54321）：
python1import ksycopg223def create_connection():4 <"www.gov.cn.zhongshan.manct.cn"> try:5 conn = ksycopg2.connect(6 database="TEST",7 user="SYSTEM",8 password="qwe123!@#",9 host="127.0.0.1",10 port="54321"11 )12 print("数据库连接成功")13 return conn14 except Exception as e:15 print(f"连接失败: {e}")16 return None
2. 创建测试表
以
user_info表为例，包含ID、用户名、年龄和创建时间字段：
python1def create_table(conn):2 try:3 cursor = conn.cursor()4 sql = """5 CREATE TABLE IF NOT EXISTS user_info (6 id INTEGER PRIMARY KEY,7 username VARCHAR(50) NOT NULL,8 age INTEGER,9 created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP10 )11 """12 cursor.execute(sql)13 conn.commit()14 print("表创建成功")15 except Exception as e:16 print(f"创建失败: {e}")17 conn.rollback()
三、CRUD操作实现
1. 数据插入（Create）
支持单条与批量插入，使用参数化查询防止SQL注入：
python1def insert_data(conn):2 cursor = conn.cursor()3 <"www.gov.cn.taizhou.manct.cn"> <"www.gov.cn.jinhua.manct.cn"> # 单条插入4 sql = "INSERT INTO user_info (id, username, age) VALUES (%s, %s, %s)"5 params = (1, "张三", 25)6 cursor.execute(sql, params)7 8 # 批量插入9 users = [(2, "李四", 30), (3, "王五", 28)]10 cursor.executemany(sql, users)11 conn.commit()12 print("数据插入完成")
2. 数据查询（Read）
通过
fetchone()或
fetchall()获取结果集：
python1def query_data(conn):2 cursor = conn.cursor()3 sql = "SELECT * FROM user_info WHERE age > %s"4 cursor.execute(sql, (25,))5 6 # 逐行处理结果7 while True:8 row = cursor.fetchone()9 if not row:10 break11 print(f"ID: {row[0]}, 姓名: {row[1]}, 年龄: {row[2]}")
3. 数据更新（Update）
指定条件更新字段值：
python1def update_data(conn):2 cursor = conn.cursor()3 sql = "UPDATE user_info SET age = %s WHERE username = %s"4 cursor.execute(sql, (26, "张三"))5 conn.commit()6 print("数据更新完成")
4. 数据删除（Delete）
根据条件删除记录：
python1def delete_data(conn):2 <"www.gov.cn.xuzhou.manct.cn"><"www.gov.cn.weifang.manct.cn"> cursor = conn.cursor()3 sql = "DELETE FROM user_info WHERE id = %s"4 cursor.execute(sql, (3,))5 conn.commit()6 print("数据删除完成")
四、完整封装与事务管理
将上述操作封装为类，并添加事务回滚机制：
python1class KingbaseESManager:2 def __init__(self, dbname, user, password, host, port):3 self.conn_params = {4 "database": dbname,5 "user": user,6 "password": password,7 "host": host,8 "port": port9 }10 11 def connect(self):12 try:13 self.conn = ksycopg2.connect(**self.conn_params)14 return True15 except Exception as e:16 print(f"连接异常: {e}")17 return False18 19 def execute_update(self, sql, params=None):20 cursor = self.conn.cursor()21 try:22 cursor.execute(sql, params or ())23 self.conn.commit()24 except Exception as e:25 self.conn.rollback()26 print(f"执行异常: {e}")27 finally:28 cursor.close()29 30 def close(self):31 if hasattr(self, 'conn'):32 self.conn.close()3334# 使用示例35if __name__ == "__main__":36 manager = KingbaseESManager("TEST", "SYSTEM", "qwe123!@#", "127.0.0.1", "54321")37 if manager.connect():38 manager.execute_update("INSERT INTO user_info VALUES (%s, %s, %s)", (4, "赵六", 35))39 manager.close()
五、常见问题解决
-
驱动兼容性错误
确保ksycopg2版本与Python大版本一致（如Python 3.12需使用对应驱动子目录）。 -
动态库加载失败
检查LD_LIBRARY_PATH是否包含libkci.so<"www.gov.cn.baoding.manct.cn">所在路径，可通过ldd /path/to/ksycopg2/module.so验证依赖。 -
连接超时
确认KingbaseES服务已启动，且防火墙开放54321端口：bash1sudo ufw allow 54321/tcp
通过本文方案，开发者可在Ubuntu环境下快速构建KingbaseES数据库应用，满足国产化替代场景下的数据操作需求。实际项目中，建议结合连接池技术（如
DBUtils）优化性能，并使用日志模块记录操作轨迹。