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

3. 创建数据表

在执行增删改查前，需确保测试表存在。这里以 user_info 表为例。

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)

4. 实现增删改查功能

4.1 新增数据

支持单条插入和批量插入。注意参数化查询可防止 SQL 注入。

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()

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():
        # 查询所有用户信息
        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 查询特定用户
        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)

        # 根据年龄范围查询
        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():
        # 更新用户年龄
        update_sql = "UPDATE user_info SET age = %s WHERE id = %s"
        db_manager.execute_update(update_sql, (26, 1))

        # 验证更新结果
        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()

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 的用户
        delete_sql = "DELETE FROM user_info WHERE id = %s"
        db_manager.execute_update(delete_sql, (5,))

        # 验证删除结果
        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()

5. 封装管理类以便复用

为了减少重复代码，可以将上述逻辑封装为一个 KingbaseESManager 类。该类统一管理连接、游标及事务提交。

import ksycopg2
from datetime import datetime

class KingbaseESManager:
    def __init__(self, dbname, user, password, host, port):
        self.conn = None
        self.db_params = {
            "database": dbname,
            "user": user,
            "password": password,
            "host": host,
            "port": port
        }

    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()

6. 总结

在 Ubuntu 上通过 Python 操作 KingbaseES 主要涉及三个环节：驱动适配、环境配置与业务封装。安装时务必注意 Python 大版本的一致性，同时正确设置 LD_LIBRARY_PATH 以避免动态链接库加载错误。通过封装通用的管理类，可以有效降低维护成本，提升代码的可读性与复用性。