Python 语言连接并操作 KWDB 数据库

前言

纯命令行操作数据还是非常的麻烦的，且没有办法进行各类批量操作，那么本文就围绕着 Python 语言对 KWDB 数据库做链接以及 CRUD 操作做一个具体的讲解，操作的前提是环境，故而还需要对环境做出妥当的部署，否则是无法正常运行我们的代码的。

环境说明

在官方网站中看到了 Python 给出了两种库操作，分别是：Psycopg 2 与 Psycopg 3，这里我使用 Psycopg 2 作为测试的库，因为我在安装 Psycopg 3 的时候需要了有些辅助库无法安装的情况，但是使用 Psycopg 2 就很顺畅，那么接下来我先罗列一下环境的概述与环境需求：

Psycopg 是专为 Python 编程语言而设计。Psycopg 完全遵循 Python DB API 2.0 规范，支持线程安全，允许多个线程共享同一连接，特别适合高并发和多线程的应用场景。

KWDB 支持用户通过 Psycopg 2 连接数据库，并执行创建、插入和查询操作。本示例演示了如何通过 Psycopg 2 驱动连接和使用 KWDB。

Python 3.10 环境

本示例使用的 Python 版本为 Python 3.10。

我这里是 python 3.13.2 版本，所以基础内容是没有问题的。

安装 Psycopg 2

pip3 install psycopg2-binary

安装完毕后直接 pip3 list 查看是否安装成功。

开启开发工具，我这里使用的是 VSCode。

基础连接代码：

con = psycopg2.connect(database="ts_db", user="teacher", password="Abcd1234", host="8.147.135.144",port="26257")print("连接成功！")
con.set_session(autocommit=True)
cur = con.cursor()

运行 sql 的方式，这里的 sql 是字符串格式的，或者直接写上就行。

cur.execute(sql)

测试代码：

#!/usr/bin/env python3# -*- coding: UTF-8 -*-import psycopg2def main():    try:
        con = psycopg2.connect(database="ts_db", user="teacher", password="Abcd1234", host="8.147.135.144",port="26257")        print("连接成功！")
        con.set_session(autocommit=True)
        cur = con.cursor()    except psycopg2.Error as e:        print(f"连接Kaiwudb失败: {e}")
    sql = "CREATE TABLE mytest_table \
        (k_timestamp timestamp NOT NULL, \
        voltage double, \
        current double, \
        temperature double \
        ) TAGS ( \
        number int NOT NULL) \
        PRIMARY TAGS(number) \
        ACTIVETIME 3h"
    try:
        cur.execute(sql)    except psycopg2.Error as e:        print(f"创建表失败: {e}")
    sql = "INSERT INTO mytest_table  \
        VALUES ('2024-07-01 10:00:00', 220.0, 3.0, 20.5,123)"
    try:
        cur.execute(sql)    except psycopg2.Error as e:        print(f"插入数据失败: {e}")
    sql = "SELECT * from mytest_table"
    try:
        cur.execute(sql)
        rows = cur.fetchall()        for row in rows:            print(f"k_timestamp: {row[0]}, voltage: {row[1]}, current: {row[2]}, temperature: {row[3]}, number: {row[4]}")    except psycopg2.Error as e:        print(f"查询数据失败: {e}")
    cur.close()
    con.close()    returnif __name__ == "__main__":
    main()

我用的是的 VSCode:

建库建表之前都操作过了，所以这里没有建表成功。可以看到查询是没有问题的，不影响全程的操作。

测试批量添加操作

这里我添加了 50 条插入的循环操作，我计算了总时间以及平均插入时间。

#!/usr/bin/env python3# -*- coding: UTF-8 -*-# 在文件顶部添加time模块导入import psycopg2import time  # 新增导入def main():    try:
        con = psycopg2.connect(database="ts_db", user="teacher", password="Abcd1234", host="8.147.135.144",port="26257")        print("连接成功！")
        con.set_session(autocommit=True)
        cur = con.cursor()    except psycopg2.Error as e:        print(f"连接Kaiwudb失败: {e}")    # 生成50条测试数据（时间戳间隔1分钟）
    base_time = '2024-07-01 10:00:00'
    data = [
        (            f'2024-07-01 10:{i:02d}:00',            220.0 + i*0.1,              
            3.0 + i*0.02,               
            20.5 + i*0.3,               
            123 + i                     
        ) 
        for i in range(50)
    ]
    sql = "INSERT INTO mytest_table VALUES (%s, %s, %s, %s, %s)"
    try:
        start_time = time.time()  # 记录开始时间
        cur.executemany(sql, data)
        end_time = time.time()    # 记录结束时间
        
        # 计算并输出耗时
        total_time = (end_time - start_time) * 1000  # 转换为毫秒
        avg_time = total_time / 50
        print(f"成功插入50条数据，总耗时：{total_time:.4f} 毫秒，平均每条耗时：{avg_time:.4f} 毫秒")    except psycopg2.Error as e:        print(f"批量插入数据失败: {e}")
    sql = "SELECT * from mytest_table"
    try:
        cur.execute(sql)
        rows = cur.fetchall()        for row in rows:            print(f"k_timestamp: {row[0]}, voltage: {row[1]}, current: {row[2]}, temperature: {row[3]}, number: {row[4]}")    except psycopg2.Error as e:        print(f"查询数据失败: {e}")
    cur.close()
    con.close()    returnif __name__ == "__main__":
    main()