轻量数据持久化 shelve | sqlite3

轻量数据持久化

shelve

shelve — Python 对象持久化 — Python 3.13.0 文档

键值对象持久化, 会生成 user.bak, user.dir, user.dat 三个文件以持久化数据

不过并不支持并发写入, 但是可以并发读取

import shelve

# 打开一个数据存储文件,
with shelve.open('user') as db:
    db['username'] = 'tiam'
    db['password'] = '123456'
    print(dict(db))  # {'password': '123456', 'username': 'tiam'}

应用场景demo

import shelve
import time


def process_data(data):
    time.sleep(1)
    print(data)


def main():
    used = shelve.open('used')
    # 假设这里有一个很长的列表耗时任务需要处理(可能中途程序会中断),
    # 避免下次重复处理, 使用shelve记录已处理的数据
    for asc in range(65, 91):
        c = chr(asc)
        # 如果已经处理过, 则跳过
        if c in used:
            continue
        try:
            process_data(c)
            print(f"Process {c} done")
            used[c] = True  # 标记已被使用
        except Exception as e:
            print(f"Process {c} failed")
        except KeyboardInterrupt:
            print("KeyboardInterrupt detected")
            break
    # 关闭shelve
    used.close()


if __name__ == '__main__':
    main()

sqlite3

官网: SQLite Home Page

介绍: SQLite 是一个C语言库，它可以提供一种轻量级的基于磁盘的数据库，这种数据库不需要独立的服务器进程，也允许需要使用一种非标准的 SQL 查询语言来访问它。

它并不需要像mysql, mongodb等数据库一样要使用地址端口去连接服务器进程使用; 它使用本地文件作为数据库持久化, 所以并不需要安装什么, 可以直接使用.

文档: sqlite3 — SQLite 数据库的 DB-API 2.0 接口 — Python 3.13.0 文档

简单的使用, 主要还是要会SQL语法

import sqlite3

conn = sqlite3.connect('demo.db')
cursor = conn.cursor()

# 执行sql语句建表
# cursor.execute("CREATE TABLE user(id, username, password)")

users = [(i, 'admin', '123456') for i in range(5)]
# 单条插入
# for user in users:
#     cursor.execute('INSERT INTO user(id, username, password) VALUES (?, ?,  ?)', user)

# 批量插入
cursor.executemany('INSERT INTO user(id, username, password) VALUES (?, ?,  ?)', users)
conn.commit()

# 查询记录
res = cursor.execute("SELECT * FROM user")
for row in res.fetchall():
    print(row)

conn.close()