经df处理的测试结果文件Dash实时读写最优存储方案选型

对于你这个单文件存储、单进程写、高频读、无额外服务部署需求的场景，SQLite3是远优于整文件替换Excel的选择，核心优势刚好匹配你的痛点：

• 彻底解决读写冲突问题：开启WAL模式后，SQLite支持读操作不阻塞写、写操作不阻塞读，完全不会出现原有方案里「Dash读文件的瞬间刚好赶上Excel被覆盖，读到半截损坏文件直接报错」的问题，应用可以持续稳定运行
• 写入开销极低：原有方案每次更新单行就要重写整个2万行的Excel文件，大量磁盘IO都浪费在重复写入未改动的数据上；SQLite单行更新只会修改对应的数据页，写入开销不到原方案的1%，哪怕每秒更新几次都没有压力
• 查询效率更高：2万条数据量级下，配合索引的SQLite查询速度比读全量Excel再用pandas筛选快10~100倍，Dash做实时筛选、状态统计的响应延迟会明显降低
• 完全符合单文件架构要求：整个数据库就是单个.db文件，不需要额外部署任何数据库服务，运维成本和你原来存Excel没有区别，备份、迁移都是直接复制文件就行

1. 存量数据初始化 #

先把你现有的Excel测试结果导入SQLite，给后续读写打好基础，核心是要设置主键、给常用查询字段加索引：

import sqlite3
import pandas as pd

# 连接数据库，文件不存在会自动创建
conn = sqlite3.connect("test_result.db")
# 读取现有Excel存量数据
df = pd.read_excel("test_result.xlsx")
# 写入数据库，首次导入用if_exists='replace'，后续运行不要加这个参数
df.to_sql("test_result", conn, if_exists="replace", index=False)
# 给高频查询字段加索引，根据你实际的筛选维度调整即可
conn.execute("CREATE INDEX IF NOT EXISTS idx_case_id ON test_result(case_id)") # 用例唯一ID作为更新、查询的核心维度
conn.execute("CREATE INDEX IF NOT EXISTS idx_status ON test_result(status)") # 测试状态是实时展示的核心筛选字段
conn.execute("CREATE INDEX IF NOT EXISTS idx_run_time ON test_result(run_time)")
conn.commit()
conn.close()

2. 必做配置：开启WAL模式 #

这一步是实现读写不冲突的核心，每次建立数据库连接后先执行这两句配置：

conn = sqlite3.connect("test_result.db")
# 开启预写日志模式，实现读写并发
conn.execute("PRAGMA journal_mode=WAL;")
# 调整同步级别，平衡数据可靠性和写入速度，测试场景下完全不会丢数据
conn.execute("PRAGMA synchronous=NORMAL;")

3. 测试进程端：替换原整文件写逻辑 #

把原来每次更新就重写整个Excel的逻辑，改成单行Update操作，每次操作新建连接、用完即关，不要长时间持有连接：

def update_single_case(case_id, status, end_time=None, error_msg=None):
    conn = sqlite3.connect("test_result.db")
    conn.execute("PRAGMA journal_mode=WAL;")
    # 按用例ID主键更新对应行，只改需要更新的字段
    conn.execute(
        """
        UPDATE test_result 
        SET status = ?, end_time = ?, error_msg = ?
        WHERE case_id = ?
        """,
        (status, end_time, error_msg, case_id)
    )
    conn.commit()
    conn.close()

注意：写操作不要加多余的耗时逻辑，更新完立刻提交事务关闭连接，避免长时间占用写锁。

4. Dash应用端：读取逻辑适配 #

Dash端不需要做复杂改造，需要取数的时候直接查询SQLite即可，读操作不会被写操作阻塞：

# 示例：拉取测试结果供前端展示
def fetch_latest_data(filter_status=None):
    conn = sqlite3.connect("test_result.db")
    conn.execute("PRAGMA journal_mode=WAL;")
    if filter_status:
        df = pd.read_sql(
            "SELECT * FROM test_result WHERE status = ? ORDER BY run_time DESC",
            conn,
            params=(filter_status,)
        )
    else:
        df = pd.read_sql("SELECT * FROM test_result ORDER BY run_time DESC", conn)
    conn.close()
    return df

如果要做实时自动刷新，直接在Dash布局里加dcc.Interval组件，设置1~5秒的间隔触发回调拉取最新数据即可，这个数据量级下完全不会有性能问题。

• 不要全局共享一个sqlite3连接对象，SQLite的连接不是线程安全的，每次读写操作新建连接、用完关闭即可，新建连接的开销极小，不会影响性能
• 不要在写事务里做网络请求、数据计算这类耗时操作，尽量缩短事务持有时间
• 2万条数据量级下基本不需要做额外优化，每隔几个月可以执行一次VACUUM命令清理文件碎片即可
• 备份直接复制整个.db文件就可以，不需要停服务，WAL模式下支持热备份

内容的提问来源于stack exchange，提问作者a-DA-v