Python实现MQTT数据写入SQLite3失败问题排查请求

核心原因 #

SQLite默认采用手动事务提交机制，脚本中执行cur.execute()后未调用con.commit()，导致插入操作仅停留在内存中，未写入数据库。此外代码还存在模块缺失和SQL拼接风险问题。

修复步骤 #

1. 补全缺失的模块导入 #

代码中使用了datetime和sys模块但未导入，会导致运行报错，需在开头添加：

import datetime as dt
import sys

2. 添加事务提交操作 #

在on_message回调中，执行完SQL插入语句后立即提交事务：

def on_message(client, userdata, msg):
    if msg.topic == "end":
        cur.close()
        con.close()
        end_msg = "Test Finished at: " + str(dt.now().strftime("%d/%m/%Y-%H:%M:%S"))
        client.publish("Important", end_msg)
        sys.exit()
    else:
        payload_data = str(msg.payload,"utf-8")
        payload_data_list = payload_data.strip("\r").split(",")
        sqlite3_command = generate_sql(test_type, "insert", payload_data_list)
        cur.execute(sqlite3_command)
        # 提交事务，将插入操作写入数据库
        con.commit()
        print(con.total_changes)

3. 改用参数化SQL查询（推荐） #

原代码通过字符串拼接生成SQL语句，存在SQL注入风险，且若Payload数据格式异常（如含空格、特殊字符）会导致SQL执行失败。修改generate_sql函数的插入语句生成逻辑：

def generate_sql(test, function, data=None):
    """Generates strings of SQL to be executed by the cursor object
       test: the type of test the script is recording as a string
       function: either the string "create" or "insert" depending on what sql needs to be generated
       return: A string of sql to be executed
    """
    if test == "big_bertha":
        if function == "create":
            create_statement = "CREATE TABLE " + test_name + " (ID INTEGER PRIMARY KEY AUTOINCREMENT, " \
                                "time DATETIME DEFAULT(STRFTIME('%Y-%m-%d %H:%M%f', 'NOW', 'localtime')), " \
                                "jog_setpoint REAL, position REAL, force_setpoint INTEGER, force INTEGER, " \
                                "pressure_t INTEGER, pressure_c INTEGER)"
            return create_statement
        else:
            # 使用?作为参数占位符，避免字符串拼接问题
            insert_statement = f"INSERT INTO {test_name} (jog_setpoint,position,force_setpoint,force,pressure_t,pressure_c) VALUES(?,?,?,?,?,?)"
            return insert_statement

对应的on_message中执行SQL的代码改为：

sqlite3_command = generate_sql(test_type, "insert")
# 将数据列表作为参数传入execute，自动处理类型转换
cur.execute(sqlite3_command, payload_data_list)
con.commit()

验证方法 #

修复后重新运行脚本，发送MQTT Payload数据后，可通过SQLite工具打开testlab.db，检查test1表是否有新数据插入。

内容的提问来源于stack exchange，提问作者bo-z