Merge pull request #27 from dimagi/gh/support-expressions

Support django expressions in AuditingQuerySet.update

field_audit/models.py

@@ -5,6 +5,7 @@
 
 from django.conf import settings
 from django.db import models, transaction
+from django.db.models import Expression
 
 from .const import BOOTSTRAP_BATCH_SIZE
 from .utils import class_import_helper
@@ -679,6 +680,9 @@ def update(self, *, audit_action=AuditAction.RAISE, **kw):
         the queryset, a fetch of audited values for the matched records is
         performed, resulting in one fetch of the current values, one update of
         the matched records, and one bulk creation of audit events.
+
+        NOTE: if django.db.models.Expressions are provided as update arguments,
+        an additional fetch of records is performed to obtain new values.
         """
         if audit_action is AuditAction.IGNORE:
             return super().update(**kw)
@@ -692,6 +696,8 @@ def update(self, *, audit_action=AuditAction.RAISE, **kw):
             return super().update(**kw)
 
         new_values = {field: kw[field] for field in fields_to_audit}
+        uses_expressions = any(
+            [isinstance(val, Expression) for val in new_values.values()])
 
         old_values = {}
         values_to_fetch = fields_to_update | {"pk"}
@@ -701,14 +707,23 @@ def update(self, *, audit_action=AuditAction.RAISE, **kw):
 
         with transaction.atomic(using=self.db):
             rows = super().update(**kw)
+            if uses_expressions:
+                # fetch updated values to ensure audit event deltas are accurate
+                # after update is performed with expressions
+                new_values = {}
+                for value in self.values(*values_to_fetch):
+                    pk = value.pop('pk')
+                    new_values[pk] = value
+            else:
+                new_values = {pk: new_values for pk in old_values.keys()}
+
             # create and write the audit events _after_ the update succeeds
             from .field_audit import request
             request = request.get()
             audit_events = []
-
             for pk, old_values_for_pk in old_values.items():
                 audit_event = AuditEvent.make_audit_event_from_values(
-                    old_values_for_pk, new_values, pk, self.model, request
+                    old_values_for_pk, new_values[pk], pk, self.model, request
                 )
                 if audit_event:
                     audit_events.append(audit_event)

tests/models.py

@@ -25,7 +25,7 @@ class SimpleModel(Model):
 @audit_fields("id", "value", audit_special_queryset_writes=True)
 class ModelWithAuditingManager(Model):
     id = AutoField(primary_key=True)
-    value = CharField(max_length=8, null=True)
+    value = CharField(max_length=16, null=True)
     non_audited_field = CharField(max_length=12, null=True)
     objects = AuditingManager()
 

tests/test_models.py

@@ -6,6 +6,7 @@
 import django
 from django.conf import settings
 from django.db import connection, models, transaction
+from django.db.models import Case, When, Value
 from django.db.utils import IntegrityError, DatabaseError
 from django.test import TestCase, override_settings
 
@@ -1116,6 +1117,33 @@ def test_update_audit_action_audit_rolls_back_if_fails(self):
         instance = ModelWithAuditingManager.objects.get(id=0)
         self.assertEqual("initial", instance.value)
 
+    def test_update_audit_action_audit_with_expressions_succeeds(self):
+        update_kwargs = {}
+        when_statements = []
+        field = ModelWithAuditingManager._meta.get_field('value')
+        for pkey in range(2):
+            obj = ModelWithAuditingManager.objects.create(id=pkey,
+                                                          value="initial")
+            attr = Value(f'updated-{pkey}', output_field=field)
+            when_statements.append(When(pk=obj.pk, then=attr))
+        case_statement = Case(*when_statements, output_field=field)
+        update_kwargs[field.attname] = case_statement
+
+        queryset = ModelWithAuditingManager.objects.all()
+        queryset.update(**dict(update_kwargs, audit_action=AuditAction.AUDIT))
+
+        instances = ModelWithAuditingManager.objects.all()
+        for instance in instances:
+            self.assertEqual(f"updated-{instance.id}", instance.value)
+            event, = AuditEvent.objects.filter(object_pk=instance.pk,
+                                               is_create=False, is_delete=False)
+            self.assertEqual(
+                "tests.models.ModelWithAuditingManager",
+                event.object_class_path)
+            self.assertEqual(
+                {"value": {"old": "initial", "new": f"updated-{instance.id}"}},
+                event.delta)
+
 
 class TestAuditingQuerySetDelete(TestCase):