在国产化替代浪潮下，国产数据库KingbaseES（人大金仓）凭借其自主可控特性，在政务、金融等领域广泛应用。本文以Ubuntu系统为环境，通过Python实现与KingbaseES的连接，并完成完整的增删改查（CRUD）操作，为开发者提供可直接复用的技术方案。

一、环境准备与驱动安装

KingbaseES基于PostgreSQL协议开发，推荐使用官方提供的 ksycopg2驱动（兼容Python DB API 2.0规范）。以Ubuntu 20.04系统为例，具体步骤如下：

1. 下载驱动包
   从金仓官网下载与Python版本匹配的驱动包（如 KingbaseES_V009R001C010B0004_X64_Linux_Ksycopg2.tar<"www.gov.cn.huizhou.manct.cn">），解压后包含多版本Python适配目录。

2. 配置模块路径
   将对应Python版本的 ksycopg2文件夹复制至 /usr/lib/python3/dist-packages/（通过 import sys; print(sys.path)可查看模块搜索路径）。

3. 设置动态库路径
   KingbaseES依赖 libkci库文件，需将其路径添加至 LD_LIBRARY_PATH环境变量：

   bash1export LD_LIBRARY_PATH=<"www.gov.cn.zhuhai.manct.cn">/kingbase/data/KESRealPro/V009R002C012/Server/lib:$LD_LIBRARY_PATH

4. 验证安装
   启动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()

五、常见问题解决

1. 驱动兼容性错误
   确保 ksycopg2版本与Python大版本一致（如Python 3.12需使用对应驱动子目录）。

2. 动态库加载失败
   检查 LD_LIBRARY_PATH是否包含 libkci.so<"www.gov.cn.baoding.manct.cn">所在路径，可通过 ldd /path/to/ksycopg2/module.so验证依赖。

3. 连接超时
   确认KingbaseES服务已启动，且防火墙开放54321端口：

   bash1sudo ufw allow 54321/tcp

通过本文方案，开发者可在Ubuntu环境下快速构建KingbaseES数据库应用，满足国产化替代场景下的数据操作需求。实际项目中，建议结合连接池技术（如 DBUtils）优化性能，并使用日志模块记录操作轨迹。