Skip to content

Commit

Permalink
Record delete snapshots (#10)
Browse files Browse the repository at this point in the history
AStoerkel1 authored May 20, 2024

Verified

This commit was created on GitHub.com and signed with GitHub’s verified signature.
1 parent a415da9 commit 7254399
Showing 5 changed files with 30 additions and 15 deletions.
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,3 +1,9 @@
## 3.4.8

* Removed `trigger_type.snapshot` so that delete triggers will now record snapshots.
Added `trigger_type.snapshot_of` to determine snapshot of OLD or NEW object.


## 3.4.7

* Added a `HISTORY_IGNORE_MODELS` setting to ignore individual models. This should be a
26 changes: 17 additions & 9 deletions history/backends/postgres.py
Original file line number Diff line number Diff line change
@@ -10,7 +10,8 @@
_ctid integer := TG_ARGV[0]::integer;
_pk_name text := TG_ARGV[1];
_record_snap boolean := TG_ARGV[2]::boolean;
_fields text[] := TG_ARGV[3:];
_snap_of text := TG_ARGV[3];
_fields text[] := TG_ARGV[4:];
_old jsonb := to_jsonb(OLD);
_new jsonb := to_jsonb(NEW);
_snapshot jsonb;
@@ -22,9 +23,15 @@
END IF;
IF _record_snap THEN
SELECT jsonb_object_agg(key, value) INTO _snapshot
FROM jsonb_each(_new)
WHERE key = ANY(_fields);
IF _snap_of = 'OLD' THEN
SELECT jsonb_object_agg(key, value) INTO _snapshot
FROM jsonb_each(_old)
WHERE key = ANY(_fields);
ELSEIF _snap_of = 'NEW' THEN
SELECT jsonb_object_agg(key, value) INTO _snapshot
FROM jsonb_each(_new)
WHERE key = ANY(_fields);
END IF;
END IF;
IF (TG_OP = 'UPDATE') THEN
@@ -133,17 +140,18 @@ def create_trigger(self, model, trigger_type):
return tr_name, []
self.execute(
"""
CREATE TRIGGER {tr_name} AFTER {trans_type} ON {table}
FOR EACH ROW EXECUTE PROCEDURE
history_record({ctid}, '{pk_col}', {snapshots}{field_list});
CREATE TRIGGER {tr_name} AFTER {trans_type} ON {table}
FOR EACH ROW EXECUTE PROCEDURE
history_record({ctid}, '{pk_col}', {snapshots}, '{snap_of}', {field_list});
""".format(
tr_name=tr_name,
trans_type=trigger_type.name.upper(),
table=model._meta.db_table,
ctid=ct.pk,
pk_col=model._meta.pk.column,
snapshots=int(conf.SNAPSHOTS and trigger_type.snapshot),
field_list=", '" + "', '".join(field_names) + "'",
snapshots=int(conf.SNAPSHOTS),
snap_of=trigger_type.snapshot_of,
field_list="'" + "', '".join(field_names) + "'",
)
)
return tr_name, field_names
4 changes: 2 additions & 2 deletions history/backends/sqlite.py
Original file line number Diff line number Diff line change
@@ -66,9 +66,9 @@ def _json_snapshot(self, fields, trigger_type):
Returns an SQL fragment that builds a JSON object from the specified model
fields.
"""
if not conf.SNAPSHOTS or not trigger_type.snapshot:
if not conf.SNAPSHOTS:
return "NULL"
return self._json_object(fields, "NEW")
return self._json_object(fields, trigger_type.snapshot_of)

def _json_changes(self, fields, trigger_type):
"""
4 changes: 2 additions & 2 deletions history/models.py
Original file line number Diff line number Diff line change
@@ -13,8 +13,8 @@ class TriggerType(models.TextChoices):
UPDATE = "U", _("Update")

@property
def snapshot(self):
return self in (TriggerType.INSERT, TriggerType.UPDATE)
def snapshot_of(self):
return "OLD" if self == TriggerType.DELETE else "NEW"

@property
def changes(self):
5 changes: 3 additions & 2 deletions tests/custom/tests.py
Original file line number Diff line number Diff line change
@@ -84,7 +84,7 @@ def test_basics(self):
# Check delete history.
self.assertEqual(delete.session_id, session.session_id)
self.assertEqual(delete.get_user(), "nobody")
self.assertIsNone(delete.snapshot)
self.assertEqual(delete.snapshot, {"id": pk, "name": "Somebody"})
self.assertIsNone(delete.changes)

def test_no_session(self):
@@ -137,7 +137,8 @@ def test_data_types(self):
self.assertEqual(update.changes["data"][0], data)
self.assertEqual(update.changes["data"][1], replace(data, answer=420))
self.assertEqual(uuid.UUID(update.snapshot["ident"]), obj.ident)
self.assertIsNone(delete.snapshot)
self.assertEqual(delete.snapshot["data"], replace(data, answer=420))
self.assertEqual(uuid.UUID(delete.snapshot["ident"]), obj.ident)
self.assertIsNone(delete.changes)

def test_change_pk(self):

0 comments on commit 7254399

Please sign in to comment.