Python脚本执行间持久化数据：有无更简便的缓存序列方法？

问题背景 #

我想了解Python是否具备简便的方法来缓存可更新的数值序列，使得脚本每次运行时都能向序列中添加新值，且下次运行时能保留这些数据。例如，我有一个由datetime和float类型组成的元组列表，其中datetime代表风速仪的记录时间，float代表记录的风速。

我最初用pickle实现，代码如下：

import os
import pickle
import datetime

db_path = "speeds.p"

# get all our previous speeds
speeds = []
if os.path.exists(db_path):
    with open(db_path, "rb") as f:
        speeds = pickle.load(f)


def data_from_endpoint():
    data = (
        (datetime.datetime(2022, 10, 22, 21, 15), 13),
        (datetime.datetime(2022, 10, 22, 21, 30), 24),
        (datetime.datetime(2022, 10, 22, 21, 45), 37)
    )

    for i in data:
        yield i


try:
    # add new speeds
    for t, v in data_from_endpoint():
        if len(speeds) == 0 or t > speeds[-1][0]:
            print(f"Adding {t}, {v}")
            speeds.append((t, v))
finally:
    # save all speeds
    with open(db_path, "wb") as f:
        pickle.dump(speeds, f)

print(f"Number of values: {len(speeds)}")

后来改用sqlite，但两种方案代码量都不小，想找更简单的实现方式。

简便实现方案 #

1. 使用标准库 shelve（推荐，无需额外安装） #

shelve是Python内置的键值对存储工具，基于pickle实现，省去了手动处理文件存在性判断、IO异常的代码，用法更简洁。

示例代码：

import shelve
import datetime

def data_from_endpoint():
    data = (
        (datetime.datetime(2022, 10, 22, 21, 15), 13),
        (datetime.datetime(2022, 10, 22, 21, 30), 24),
        (datetime.datetime(2022, 10, 22, 21, 45), 37)
    )
    for i in data:
        yield i

# 打开shelve数据库，自动处理文件创建/读取
with shelve.open('speeds_db') as db:
    # 获取已有数据，默认返回空列表
    speeds = db.get('speeds', [])
    # 添加新数据（保留时间戳递增逻辑）
    for t, v in data_from_endpoint():
        if not speeds or t > speeds[-1][0]:
            print(f"Adding {t}, {v}")
            speeds.append((t, v))
    # 保存更新后的数据
    db['speeds'] = speeds
    print(f"Number of values: {len(speeds)}")

核心优势：

• 无需额外安装依赖，开箱即用
• 自动处理文件IO流程，代码量比手动pickle减少近一半
• 支持存储大部分Python原生对象（包括datetime、元组等）

2. 使用 pandas 快速序列化（适合数据分析场景） #

如果你的项目已经依赖pandas，用它的序列化工具可以进一步简化代码，同时还能直接利用pandas的数据处理能力。

示例代码：

import pandas as pd
import datetime

def data_from_endpoint():
    data = (
        (datetime.datetime(2022, 10, 22, 21, 15), 13),
        (datetime.datetime(2022, 10, 22, 21, 30), 24),
        (datetime.datetime(2022, 10, 22, 21, 45), 37)
    )
    return pd.DataFrame(data, columns=['timestamp', 'speed'])

# 读取已有数据，不存在则创建空DataFrame
try:
    df = pd.read_pickle('speeds_df.pkl')
except FileNotFoundError:
    df = pd.DataFrame(columns=['timestamp', 'speed'])

new_df = data_from_endpoint()
# 筛选未存在的新时间戳
new_entries = new_df[new_df['timestamp'] > df['timestamp'].max()] if not df.empty else new_df
# 合并数据并保存
df = pd.concat([df, new_entries], ignore_index=True)
df.to_pickle('speeds_df.pkl')

print(f"Number of values: {len(df)}")

核心优势：

• 适合后续需要做风速统计、可视化的场景
• 内置数据去重、筛选逻辑，无需手动写判断
• 支持更高效的序列化格式（如feather、parquet）

3. 使用 diskcache 第三方库（功能丰富的缓存工具） #

如果需要更灵活的缓存控制（比如设置过期时间、缓存大小限制），可以用diskcache，安装后用法同样简洁。

先安装依赖：

pip install diskcache

示例代码：

from diskcache import Cache
import datetime

def data_from_endpoint():
    data = (
        (datetime.datetime(2022, 10, 22, 21, 15), 13),
        (datetime.datetime(2022, 10, 22, 21, 30), 24),
        (datetime.datetime(2022, 10, 22, 21, 45), 37)
    )
    for i in data:
        yield i

# 创建缓存实例
cache = Cache('speeds_cache')
# 获取已有数据，默认空列表
speeds = cache.get('speeds', [])
# 添加新数据
for t, v in data_from_endpoint():
    if not speeds or t > speeds[-1][0]:
        print(f"Adding {t}, {v}")
        speeds.append((t, v))
# 保存更新
cache.set('speeds', speeds)
print(f"Number of values: {len(speeds)}")
# 关闭缓存连接
cache.close()

核心优势：

• 支持并发访问，适合多进程/线程场景
• 可设置缓存过期时间、自动清理策略
• 性能优于shelve，适合大规模数据存储

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