SQLite - Bulk insert compare

In progress, will expand the details out.

Create table with generated uniqueness row hash

PRAGMA journal_mode = 'wal';

DROP TABLE IF EXISTS agreements;

CREATE TABLE IF NOT EXISTS agreements (
    file_id TEXT NOT NULL,
    dag TEXT NOT NULL,
    file_code TEXT NOT NULL,
    org_name TEXT,
    uploaded datetime NOT NULL,
    expired datetime,
    resources TEXT,
	compare_id TEXT GENERATED ALWAYS AS (hex(dag || file_code || uploaded || coalesce(expired, '') || coalesce(resources, ''))) STORED,
    UNIQUE(compare_id)
);

Insert all records, updates, duplicates, etc

insert or replace into agreements (file_id, dag, file_code, uploaded, expired, org_name, resources)
values	('previous', 'org-1', 'abc', '2026-01-01', null, null, '1, 2'),
       ('previous', 'org-2', 'abc', '2026-01-01', null, null, null),
       ('previous', 'org-3', 'abc', '2026-01-01', '2026-01-03', null, null),
	   ('current', 'org-3', 'abc', '2026-01-01', '2026-01-03', null, null),
       ('current', 'org-1', 'abc', '2026-01-01', null, null, null),
       ('current', 'org-1', 'def', '2026-01-01', null, null, null),
       ('current', 'org-2', 'abc', '2026-01-01', '2026-10-01', null, null),
       ('current', 'org-4', 'def', '2026-01-01', null, null, null);

Partition records into groups take first value current if there is more than 1 value record has changed

select row_number() over (partition by dag, file_code order by file_id asc) row_num, dag, file_code, org_name, uploaded, expired, resources	
from agreements
order by dag, file_code, row_num

Sample code

def result_set(input_records: list[AgreementExtra]) -> dict[str, list[Any]]:
    added: set[str] = set()
    removed: set[str] = set(['org-4'])
    changed: set[str] = set()
    results = dict(added=[], changed=[], removed=[])
    for key, dag_records in groupby(input_records, lambda z: z.dag):
        all_records = sorted(deque(dag_records), key=lambda z: z.row_num)
        current_dag = peekable(all_records).peek().dag
        latest = sorted(filter(lambda z: z.row_num == 1, all_records), key=lambda z: z.file_code)
        if current_dag in added:
            results['added'].extend(latest)
            continue
        if current_dag in removed:
            results['removed'].extend(latest)
            continue
        if len(list(filter(lambda z: z.row_num > 1, all_records))) > 0:
            results['changed'].extend(latest)
    return results

with pydapper.connect("sqlite://pydapper.db") as commands:
    commands.execute(create_stmt)
    commands.execute("DELETE FROM agreements")
    commands.execute(insert_stmt)
    records = commands.query(sql=select_stmt, model=AgreementExtra)
    ic(result_set(input_records=records))