PythonでSQlite3に書き込む前後のdiffを取る方法

はじめに

PythonでSQlite3に書き込む前後のdiffを取る方法をご紹介します。
実施に際しては、以下の手順を踏んでいます。

1. インサート前のデータ取得
   • DBからインサート前のデータを取得し、それを一時的に保存する。
2. データのインサート
   • 指定されたデータをDBにインサートする。
3. インサート後のデータ取得
   • 再びDBからデータを取得し、インサート後の状態を保存する。
4. 比較（Diff）
   • setを使ってインサート前後のデータを比較し、レコードの追加・変更を特定する。

方法

import json
import sqlite3


def create_db_and_table(db_name):
    """データベースとテーブルを作成する関数"""
    with sqlite3.connect(db_name) as conn:
        cursor = conn.cursor()
        # TODO: hogeのところは要変更
        # created_atはローカル時間で自動登録
        cursor.execute(
            """CREATE TABLE IF NOT EXISTS hoge (
                id INTEGER PRIMARY KEY,
                url TEXT,
                created_at TIMESTAMP DEFAULT (DATETIME('now','localtime'))
            )"""
        )


def insert_data(db_name, data):
    """データをデータベースに挿入する関数"""
    with sqlite3.connect(db_name) as conn:
        cursor = conn.cursor()
        # TODO: hogeのところは要変更
        # INSERT OR IGNORE INTOで重複を登録しない
        cursor.executemany(
            """INSERT OR IGNORE INTO hoge (id, url)
            VALUES (?, ?)""",
            [(item["id"], item["url"]) for item in data],
        )


def get_data_from_db(db_name):
    """データベースから全データを取得する関数"""
    with sqlite3.connect(db_name) as conn:
        cursor = conn.cursor()
        # TODO: hogeのところは要変更
        cursor.execute("SELECT * FROM hoge")

        return cursor.fetchall()


def compare_data(before, after):
    """インサート前後のデータを比較して、差分を返す関数"""
    before_set = set(before)
    after_set = set(after)

    return after_set - before_set


def main():
    db_name = "hoge.db"

    json_data = """
    [{"id": "1", "url": "huga"},
    {"id": "2", "url": "huga"},
    {"id": "3", "url": "huga"}]
    """

    data = json.loads(json_data)

    # テーブル作成
    create_db_and_table(db_name)

    # インサート前のデータを取得
    before_insert = get_data_from_db(db_name)

    # データをインサート
    insert_data(db_name, data)

    # インサート後のデータを取得
    after_insert = get_data_from_db(db_name)

    # データの差分を取得
    diff = compare_data(before_insert, after_insert)

    print(str(diff))


if __name__ == "__main__":
    main()