sqlite-history

使用触发器跟踪SQLite表的变化

安装

使用pip安装此库

pip install sqlite-history

用法

此库可用于在SQLite数据库上配置触发器，以便任何对表的插入、更新或删除都将记录在单独的表中。

您可以使用enable_history()函数为表启用历史跟踪

import sqlite_history
import sqlite3

conn = sqlite3.connect("data.db")
conn.execute("CREATE TABLE table1 (id INTEGER PRIMARY KEY, name TEXT)")
sqlite_history.configure_history(conn, "table1")

或者，您可以使用通过python -m sqlite_history提供的CLI界面

python -m sqlite_history data.db table1 [table2 table3 ...]

使用--all为所有表配置

python -m sqlite_history data.db --all

工作原理

给定以下模式的表

CREATE TABLE people (
    id INTEGER PRIMARY KEY,
    name TEXT,
    age INTEGER,
    weight REAL
);

此库将创建一个名为_people_history的新表，其模式如下

CREATE TABLE _people_history (
    _rowid INTEGER,
   id INTEGER,
   name TEXT,
   age INTEGER,
   weight REAL,
    _version INTEGER,
    _updated INTEGER,
    _mask INTEGER
);
CREATE INDEX idx_people_history_rowid ON _people_history (_rowid);

_rowid列引用原始表中正在跟踪的行的rowid。如果一行被更新多次，则此表中将有多行具有相同的_rowid。

id、name、age和weight列表示更新时分配给行的新的值。这些也可以是null，可能表示没有更改或表示将值设置为null（因此有_mask列）。

_version列是一个单调递增的整数，每次更新一行时都会递增。

_updated列是一个显示更改记录时间的戳。它以自Unix纪元以来的毫秒存储。要在SQL查询中将它转换为可读的UTC日期，您可以使用strftime('%Y-%m-%d %H:%M:%S', _updated / 1000, 'unixepoch')

_mask列是一个位掩码，表示更新中哪些列发生了更改。位掩码是通过将以下值相加计算得出的

1: id
2: name
4: age
8: weight

具有不同模式的表将具有不同的_mask值。

一个 _mask 为 -1 的标志表示该行已被删除。

创建了以下触发器以填充 _people_history 表

CREATE TRIGGER people_insert_history
AFTER INSERT ON people
BEGIN
    INSERT INTO _people_history (_rowid, id, name, age, weight, _version, _updated, _mask)
    VALUES (new.rowid, new.id, new.name, new.age, new.weight, 1, cast((julianday('now') - 2440587.5) * 86400 * 1000 as integer), 15);
END;

CREATE TRIGGER people_update_history
AFTER UPDATE ON people
FOR EACH ROW
BEGIN
    INSERT INTO _people_history (_rowid, id, name, age, weight, _version, _updated, _mask)
    SELECT old.rowid, 
        CASE WHEN old.id != new.id then new.id else null end, 
        CASE WHEN old.name != new.name then new.name else null end, 
        CASE WHEN old.age != new.age then new.age else null end, 
        CASE WHEN old.weight != new.weight then new.weight else null end,
        (SELECT MAX(_version) FROM _people_history WHERE _rowid = old.rowid) + 1,
        cast((julianday('now') - 2440587.5) * 86400 * 1000 as integer),
        (CASE WHEN old.id != new.id then 1 else 0 end) + (CASE WHEN old.name != new.name then 2 else 0 end) + (CASE WHEN old.age != new.age then 4 else 0 end) + (CASE WHEN old.weight != new.weight then 8 else 0 end)
    WHERE old.id != new.id or old.name != new.name or old.age != new.age or old.weight != new.weight;
END;

CREATE TRIGGER people_delete_history
AFTER DELETE ON people
BEGIN
    INSERT INTO _people_history (_rowid, id, name, age, weight, _version, _updated, _mask)
    VALUES (
        old.rowid,
        old.id, old.name, old.age, old.weight,
        (SELECT COALESCE(MAX(_version), 0) from _people_history WHERE _rowid = old.rowid) + 1,
        cast((julianday('now') - 2440587.5) * 86400 * 1000 as integer),
        -1
    );
END;

开发

要为此库做出贡献，首先检出代码。然后创建一个新的虚拟环境

cd sqlite-history
python -m venv venv
source venv/bin/activate

现在安装依赖项和测试依赖项

pip install -e '.[test]'

要运行测试

pytest