python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python连接KingbaseES数据库

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

作者：正在走向自律

本文介绍了在Ubuntu系统中使用Python连接KingbaseES数据库的方法,主要内容包括:安装与Python版本匹配的ksycopg2驱动;配置环境变量和连接参数;实现数据库连接、建表及增删改查操作;封装一个可复用的数据库操作类,通过代码示例演示了数据插入、查询、更新和删除等常见操作

1.环境准备与驱动安装

KingbaseES提供了专门的Python驱动包ksycopg2，它是基于PythonDBAPI2.0规范实现的线程安全数据库适配器。为了使用ksycopg2连接KingbaseES数据库，需要按照以下步骤进行环境准备和驱动安装。

1.1 科普ksycopg2知识

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

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

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

更多关于ksycopg2的知识，请参考金仓KingbaseES官方手册。

1.2 官方下载ksycopg2驱动

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

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版本，这一点为国产金仓数据库点赞

将这些驱动文件放置在Python的模块搜索路径中。可以通过以下Python代码查看当前模块搜索路径：

import sys
print(sys.path)

1.3 安装ksycopg2驱动

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

import sys
print(sys.path)

2、根据Python模块的位置，将ksycopg2驱动文件放置在合适的目录中。例如，在Ubuntu系统中，可以将驱动文件放置在：

/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_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 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.KingbaseES数据库用户评价

许多用户在使用KingbaseES数据库时表现良好，特别是在与传统数据库如MySQL和PostgreSQL相比时，KingbaseES在性能和扩展性方面表现出色。以下是一些用户评价：

用户A：“KingbaseES数据库连接Python库ksycopg2非常稳定，连接速度很快，数据库操作也高效，非常推荐使用！”

用户B：“在 kingbasees 上运行数据库连接后，发现查询速度比预期快得多，适合大规模数据处理的场景。”

用户C：“ksycopg2驱动安装过程非常简单，完全按照教程操作后，就能轻松连接到KingbaseES数据库，非常满意！”

用户D：“KingbaseES的Python驱动ksycopg2非常强大，支持多种数据库连接方式，非常灵活。”

6.总结

本文介绍了在Ubuntu系统中使用Python连接KingbaseES数据库的方法。主要内容包括：安装与Python版本匹配的ksycopg2驱动； 配置环境变量和连接参数； 实现数据库连接、建表及增删改查操作； 封装一个可复用的数据库操作类。 通过具体代码示例演示了数据插入、查询、更新和删除等常见操作，并提供了验证方法。

通过以上步骤，开发者可以轻松地在Ubuntu服务器上使用KingbaseES数据库进行数据管理和操作。

到此这篇关于Python连接KingbaseES数据库实现增删改查（Ubuntu系统）的文章就介绍到这了,更多相关Python连接KingbaseES数据库内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:
阅读全文