Record delete snapshots

CHANGELOG.md

@@ -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

history/backends/postgres.py

@@ -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

history/backends/sqlite.py

@@ -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):
         """

history/models.py

@@ -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):

tests/custom/tests.py

@@ -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):