如何从SQLite持续获取传感器实时数据并实现打印？

你的代码仅执行了一次查询，只能拿到运行时的静态数据集。要持续获取传感器更新的实时数据，需要周期性重复执行查询，还可以优化为只读取新增数据（避免重复打印旧内容）。

修改后的代码示例（增量获取，推荐） #

import sqlite3
import time

def get_new_sensor_data(last_timestamp=None):
    # 建立数据库连接，关闭事务隔离确保读取最新数据
    conn = sqlite3.connect('sqlite2.db')
    conn.isolation_level = None
    c = conn.cursor()
    
    if last_timestamp:
        # 仅查询上次获取之后新增的数据
        c.execute('SELECT time, data FROM sensor WHERE time > ?', (last_timestamp,))
    else:
        # 首次运行获取全部数据
        c.execute('SELECT time, data FROM sensor')
    
    rows = c.fetchall()
    # 关闭游标和连接，释放资源
    c.close()
    conn.close()
    return rows

def monitor_sensor(interval=2):
    last_record_time = None
    while True:
        new_records = get_new_sensor_data(last_record_time)
        for row in new_records:
            print(f"时间: {row[0]}, 数据: {row[1]}")
            # 更新最后一条记录的时间，用于下次增量查询
            last_record_time = row[0]
        # 每隔指定秒数查询一次
        time.sleep(interval)

if __name__ == "__main__":
    monitor_sensor()

关键修改说明 #

• 循环监控：用while True实现持续运行，每隔指定时间重复查询数据库。
• 增量查询：记录最后一次获取的时间戳，只读取新增数据，避免重复打印旧内容，提升效率。
• 连接配置：设置conn.isolation_level = None关闭事务隔离，确保每次查询能读取到传感器刚写入的最新数据（SQLite默认隔离级别会缓存旧数据）。
• 资源管理：每次查询后关闭游标和连接，避免长时间占用数据库资源。

简化版（全量重复打印） #

如果不需要增量获取，只想每次打印所有数据，可使用以下简化代码：

import sqlite3
import time

def print_all_sensor_data():
    conn = sqlite3.connect('sqlite2.db')
    conn.isolation_level = None
    c = conn.cursor()
    c.execute('SELECT time, data FROM sensor')
    for row in c.fetchall():
        print(f"时间: {row[0]}, 数据: {row[1]}")
    c.close()
    conn.close()

while True:
    print_all_sensor_data()
    # 可选：清空控制台让输出更整洁
    # print("\033c", end="")
    time.sleep(2)

内容的提问来源于stack exchange，提问作者sharon