simonw/sqlite-chronicle
GitHub: simonw/sqlite-chronicle
利用 SQLite 触发器自动追踪表中行的插入、更新和删除时间与版本号的 Python 库。
Stars: 56 | Forks: 0
# sqlite-chronicle
[](https://pypi.org/project/sqlite-chronicle/)
[](https://github.com/simonw/sqlite-chronicle/releases)
[](https://github.com/simonw/sqlite-chronicle/actions?query=workflow%3ATest)
[](https://github.com/simonw/sqlite-chronicle/blob/main/LICENSE)
使用 trigger 跟踪 SQLite 表中行的插入、更新或删除时间
## 核心概念
创建一个 `_chronicle_{table_name}` 表来伴随常规表。常规表中的每一行在 chronicle 表中都有对应的行，用于记录其添加或最后更新的时间戳以及递增的版本号，所有这些都通过 SQLite trigger 来维护。这让你能够高效地识别自上次时间戳或版本以来，哪些行被添加、更新或删除了。
## 安装
```
pip install sqlite-chronicle
```
## 命令行界面
你可以使用命令行界面为 SQLite 数据库中的特定表启用 chronicle，传入一个或多个表名：
```
sqlite-chronicle database.db table_1 table_2
```
要禁用 chronicle 跟踪，请使用 `--disable` 标志：
```
sqlite-chronicle database.db table_1 --disable
```
你也可以使用 [uvx](https://docs.astral.sh/uv/concepts/tools/) 在未安装此工具的情况下直接运行它：
```
uvx sqlite-chronicle --help
```
## Python API
此包暴露了几个用于配置和使用 chronicle 表的 Python 函数：
### enable_chronicle(conn, table_name)
该模块提供了一个函数：`sqlite_chronicle.enable_chronicle(conn, table_name)`，它执行以下操作：
1. 检查 `_chronicle_{table_name}` 表是否已存在。如果存在，则什么也不做。否则...
2. 创建该表，其具有与原表相同的主键列，外加整数列 `__added_ms`、`__updated_ms`、`__version` 和 `__deleted`
3. 在 chronicle 表中为原表中的每一行创建对应的新行，将 `__added_ms` 和 `__updated_ms` 设置为当前时间戳（以毫秒为单位），并将 `__version` 列的起始值设为 1，后续行依次递增
4. 在表上设置四个 trigger：
- 一个 BEFORE INSERT trigger，它将现有行数据快照到 helper 表中，以支持 `INSERT OR REPLACE` 的变更检测
- 一个 AFTER INSERT trigger，它在 chronicle 表中创建一个新行，将 `__added_ms` 和 `__updated_ms` 设置为当前时间，并将 `__version` 设置为比该表当前最大版本号高 1。对于 `INSERT OR REPLACE`，它会与快照进行比较以检测数据是否实际发生了更改
- 一个 AFTER UPDATE trigger，它更新 `__updated_ms` 时间戳并递增 `__version` —— 但前提是该行中至少有一列发生了更改
- 一个 AFTER DELETE trigger，它更新 `__updated_ms`，递增 `__version` 并在 `deleted` 列中置为 `1` —— 但仅针对真正的删除，不包括 `INSERT OR REPLACE` 内部的隐式删除
如果表不存在，或者没有单一或复合主键，该函数将抛出 `sqlite_chronicle.ChronicleError` 异常。
请注意，一个表的 `__version` 是一个全局递增的数字，因此每次设置时，都会被设置为整个表当前的 `max(__version)` + 1。
最终生成的 chronicle 表如下所示：
| id | __added_ms | __updated_ms | __version | __deleted |
|-----|---------------|---------|--------|---------|
| 47 | 1694408890954 | 1694408890954 | 2 | 0 |
| 48 | 1694408874863 | 1694408874863 | 3 | 1 |
| 1 | 1694408825192 | 1694408825192 | 4 | 0 |
| 2 | 1694408825192 | 1694408825192 | 5 | 0 |
| 3 | 1694408825192 | 1694408825192 | 6 | 0 |
### disable_chronicle(conn, table_name)
使用 `sqlite_chronicle.disable_chronicle(conn, table_name)` 移除表的 chronicle 跟踪。这将：
1. 删除 `_chronicle_{table_name}` 表
2. 移除所有相关的 trigger
3. 移除版本索引
如果禁用了 chronicle 则返回 `True`，如果该表不存在 chronicle 表则返回 `False`。
```
import sqlite_chronicle
# 禁用 dogs 表上的 chronicle
result = sqlite_chronicle.disable_chronicle(conn, "dogs")
if result:
print("Chronicle disabled")
else:
print("No chronicle found")
```
### is_chronicle_enabled(conn, table_name)
使用 `sqlite_chronicle.is_chronicle_enabled(conn, table_name)` 检查表是否启用了 chronicle 跟踪。
如果给定表存在 chronicle 表则返回 `True`，否则返回 `False`。
```
import sqlite_chronicle
if sqlite_chronicle.is_chronicle_enabled(conn, "dogs"):
print("Chronicle is enabled for dogs")
else:
print("Chronicle is not enabled for dogs")
```
### list_chronicled_tables(conn)
使用 `sqlite_chronicle.list_chronicled_tables(conn)` 获取所有启用了 chronicle 跟踪的表的列表。
```
import sqlite_chronicle
tables = sqlite_chronicle.list_chronicled_tables(conn)
print(f"Chronicle is enabled for: {tables}")
# 输出：Chronicle 已为以下表启用：['dogs', 'cats']
```
### upgrade_chronicle(conn, table_name)
此函数检测指定表先前是否已为其创建了旧版本的 chronicle 表和 trigger，如果是，它将把该表升级到最新的实现，同时保留现有时间戳和版本数据。
### updates_since(conn, table_name, since=None, batch_size=1000)
`sqlite_chronicle.updates_since()` 函数返回一个生成器，遍历 `Change` 对象的列表。
这些对象表示自 `since` 版本号以来表中行发生的更改；如果未提供 `since`，则表示自最初以来发生的更改。
- `conn` 是一个 SQLite 连接对象
- `table_name` 是一个字符串，包含要获取更改的表名
- `since` 是一个可选的整数版本号 —— 如果未提供，则返回所有更改
- `batch_size` 是一个内部细节，控制一次从数据库返回的行数。你不需要更改它，因为该函数实现了自己的内部分页。
从生成器返回的每个 `Change` 类似于这样：
```
Change(
pks=(5,),
added_ms=1701836971223,
updated_ms=1701836971223,
version=5,
row={'id': 5, 'name': 'Simon'},
deleted=False
)
```
`Change` 是一个 dataclass，具有以下属性：
- `pks` 是该行主键值的元组 —— 对于普通主键，这是一个只有单个项的元组；对于复合主键，则包含多个项
- `added_ms` 是行被添加时的时间戳（以毫秒为单位）
- `updated_ms` 是行被最后更新时的时间戳（以毫秒为单位）
- `version` 是该行的版本号 —— 你可以将其用作 `since` 值来获取该点之后的更改
- `row` 是一个字典，包含该行的当前值 —— 如果该行已被删除，则这些值将为 `None`（主键除外）
- `deleted` 如果该行未被删除则为 `0`，如果已被删除则为 `1`
每次调用此方法时，你都应该记录下你看到的最后一个 `version` 号，这样你就可以在未来的调用中将其作为 `since` 值传入，以获取该点之后发生的更改。
请注意，如果在两次调用此函数期间某一行发生了多次更新，你仍然只会看到该行对应的一个 `Change` 对象 —— `updated_ms` 和 `version` 将反映最近一次更新。
## 实现说明
- 完整支持 `INSERT OR REPLACE` 和 `INSERT ... ON CONFLICT DO UPDATE` (UPSERT)。如果 `INSERT OR REPLACE` 向现有行写入了相同的数据，则不会发生版本提升。
- 如果某行被删除后又被重新插入相同的主键，`__added_ms` 将重置为重新插入的时间（并且 `__deleted` 翻转回 `0`）。这会将“取消删除”视为一次全新添加，而不是原始行历史的延续。
- 目前不支持对作为记录主键一部分的列进行更新。
## 潜在应用
Chronicle 表可用于高效回答“自上次检查以来，哪些行已被插入、更新或删除”的问题 —— 通过查看 `version` 列，该列具有索引，可以快速回答此问题。
这有许多潜在的应用，包括：
- 同步和复制：其他数据库可以“订阅”表，跟踪它们上次刷新副本的时间，并仅请求自上次以来发生更改的行 —— 并删除已标记为已删除的行。
- 索引：如果你需要更新 Elasticsearch 索引或向量数据库的 embedding 索引等，你可以仅针对自上次运行以来发生更改的记录执行此操作 —— 另请参阅[非规范化查询引擎设计模式](https://2017.djangocon.us/talks/the-denormalized-query-engine-design-pattern/)
- 增强：[datasette-enrichments](https://github.com/datasette/datasette-enrichments) 需要持久化类似于“每个地址列都应被地理编码”的状态信息 —— 然后让一个增强任务每 X 秒运行一次，查找新插入或更新的行并仅对它们进行增强。
- 向人们展示自上次访问以来发生了什么变化 —— 比如“自昨天以来已有 52 行被更新，16 行被删除”之类的事情。
## SQL schema 示例
这是一个 SQL schema 示例，展示了为 `dogs` 表创建的 `_chronicle_dogs` 表及其 trigger：
标签：Python, SQLite, 数据变更跟踪, 数据同步, 数据库工具, 无后门, 触发器, 逆向工具