feat: give users the option to run the json migration asyncly

auditlog/apps.py

@@ -11,3 +11,7 @@ def ready(self):
         from auditlog.registry import auditlog
 
         auditlog.register_from_settings()
+
+        from auditlog import models
+
+        models.changes_func = models._changes_func()

auditlog/conf.py

@@ -27,3 +27,11 @@
     settings, "AUDITLOG_CID_HEADER", "x-correlation-id"
 )
 settings.AUDITLOG_CID_GETTER = getattr(settings, "AUDITLOG_CID_GETTER", None)
+
+# migration
+settings.AUDITLOG_TWO_STEP_MIGRATION = getattr(
+    settings, "AUDITLOG_TWO_STEP_MIGRATION", False
+)
+settings.AUDITLOG_USE_TEXT_CHANGES_IF_JSON_IS_NOT_PRESENT = getattr(
+    settings, "AUDITLOG_USE_TEXT_CHANGES_IF_JSON_IS_NOT_PRESENT", False
+)

auditlog/management/commands/auditlogmigratejson.py

@@ -0,0 +1,108 @@
+from django.conf import settings
+from django.core.management.base import BaseCommand
+from django.core.paginator import Paginator
+
+from auditlog.models import LogEntry
+
+
+class Command(BaseCommand):
+    help = "Migrates changes from changes_text to json changes."
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "-d",
+            "--database",
+            default=None,
+            help="If provided, we will use native db operations.",
+            dest="db",
+            type=str,
+            choices=["postgres", "mysql", "oracle"],
+        )
+        parser.add_argument(
+            "-b",
+            "--bactch-size",
+            default=None,
+            help="Split the migration into multiple batches",
+            dest="batch_size",
+            type=int,
+        )
+
+    def handle(self, *args, **options):
+        database = options["db"]
+        batch_size = options["batch_size"]
+
+        if not self.check_logs():
+            return
+
+        if database:
+            result = self.migrate_using_sql(database)
+            self.stdout.write(
+                f"Updated {result} records using native database operations."
+            )
+        else:
+            result = self.migrate_using_django(batch_size)
+            self.stdout.write(f"Updated {result} records using django operations.")
+
+        self.check_logs()
+
+    def check_logs(self):
+        count = self.get_logs().count()
+        if count:
+            self.stdout.write(f"There are {count} records that needs migration.")
+            return True
+
+        self.stdout.write("All records are have been migrated.")
+        if settings.AUDITLOG_USE_TEXT_CHANGES_IF_JSON_IS_NOT_PRESENT:
+            self.stdout.write(
+                "You can now set AUDITLOG_USE_TEXT_CHANGES_IF_JSON_IS_NOT_PRESENT to False."
+            )
+
+        return False
+
+    def get_logs(self):
+        return LogEntry.objects.filter(
+            changes_text__isnull=False, changes__isnull=True
+        ).exclude(changes_text__exact="")
+
+    def migrate_using_django(self, batch_size):
+        def _apply_django_migration(_logs) -> int:
+            import json
+
+            updated = []
+            for log in _logs:
+                try:
+                    log.changes = json.loads(log.changes_text)
+                except ValueError:
+                    pass
+                else:
+                    updated.append(log)
+
+            LogEntry.objects.bulk_update(updated, fields=["changes"])
+            return len(updated)
+
+        logs = self.get_logs()
+        if not batch_size:
+            return _apply_django_migration(logs)
+
+        total_updated = 0
+        for page in Paginator(logs, batch_size):
+            total_updated += _apply_django_migration(page.object_list)
+        return total_updated
+
+    def migrate_using_sql(self, database):
+        from django.db import connection
+
+        def postgres():
+            with connection.cursor() as cursor:
+                cursor.execute(
+                    'UPDATE auditlog_logentry SET changes="changes_text"::jsonb'
+                )
+                return cursor.cursor.rowcount
+
+        if database == "postgres":
+            return postgres()
+        else:
+            self.stderr.write(
+                "Not yet implemented. Run this management commad without passing a -d/--database argument."
+            )
+            return 0

auditlog/migrations/0015_alter_logentry_changes.py

@@ -1,18 +1,45 @@
 # Generated by Django 4.0 on 2022-08-04 15:41
+from typing import List
 
+from django.conf import settings
 from django.db import migrations, models
 
 
-class Migration(migrations.Migration):
-
-    dependencies = [
-        ("auditlog", "0014_logentry_cid"),
-    ]
+def two_step_migrations() -> List:
+    if settings.AUDITLOG_TWO_STEP_MIGRATION:
+        return [
+            migrations.RenameField(
+                model_name="logentry",
+                old_name="changes",
+                new_name="changes_text",
+            ),
+            migrations.AddField(
+                model_name="logentry",
+                name="changes",
+                field=models.JSONField(null=True, verbose_name="change message"),
+            ),
+        ]
 
-    operations = [
+    return [
+        migrations.AddField(
+            model_name="logentry",
+            name="changes_text",
+            field=models.TextField(null=True, verbose_name="change message"),
+        ),
         migrations.AlterField(
             model_name="logentry",
             name="changes",
             field=models.JSONField(null=True, verbose_name="change message"),
         ),
     ]
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("auditlog", "0014_logentry_cid"),
+    ]
+
+    atomic = not settings.AUDITLOG_TWO_STEP_MIGRATION
+
+    operations = [*two_step_migrations()]

auditlog/models.py

@@ -2,7 +2,7 @@
 import json
 from copy import deepcopy
 from datetime import timezone
-from typing import Any, Dict, List, Union
+from typing import Any, Callable, Dict, List, Union
 
 from dateutil import parser
 from dateutil.tz import gettz
@@ -356,6 +356,7 @@ class Action:
     action = models.PositiveSmallIntegerField(
         choices=Action.choices, verbose_name=_("action"), db_index=True
     )
+    changes_text = models.TextField(null=True, verbose_name=_("change message"))
     changes = models.JSONField(null=True, verbose_name=_("change message"))
     actor = models.ForeignKey(
         to=settings.AUTH_USER_MODEL,
@@ -405,7 +406,7 @@ def changes_dict(self):
         """
         :return: The changes recorded in this log entry as a dictionary object.
         """
-        return self.changes or {}
+        return changes_func(self)
 
     @property
     def changes_str(self, colon=": ", arrow=" \u2192 ", separator="; "):
@@ -579,3 +580,26 @@ def bulk_related_objects(self, objs, using=DEFAULT_DB_ALIAS):
         # method.  However, because we don't want to delete these related
         # objects, we simply return an empty list.
         return []
+
+
+# should I add a signal reciver for setting_changed?
+changes_func = None
+
+
+def _changes_func() -> Callable[[LogEntry], Dict]:
+    def json_then_text(instance: LogEntry) -> Dict:
+        if instance.changes:
+            return instance.changes
+        elif instance.changes_text:
+            try:
+                return json.loads(instance.changes_text)
+            except ValueError:
+                pass
+        return {}
+
+    def default(instance: LogEntry) -> Dict:
+        return instance.changes or {}
+
+    if settings.AUDITLOG_USE_TEXT_CHANGES_IF_JSON_IS_NOT_PRESENT:
+        return json_then_text
+    return default

auditlog_tests/test_settings.py

@@ -32,7 +32,7 @@
         "NAME": os.getenv(
             "TEST_DB_NAME", "auditlog" + os.environ.get("TOX_PARALLEL_ENV", "")
         ),
-        "USER": os.getenv("TEST_DB_USER", "postgres"),
+        "USER": os.getenv("TEST_DB_USER", "aqeelat"),
         "PASSWORD": os.getenv("TEST_DB_PASS", ""),
         "HOST": os.getenv("TEST_DB_HOST", "127.0.0.1"),
         "PORT": os.getenv("TEST_DB_PORT", "5432"),

auditlog_tests/test_two_step_json_migration.py

@@ -0,0 +1,69 @@
+import json
+from io import StringIO
+
+from django.core.management import call_command
+from django.test import TestCase
+
+from auditlog.models import LogEntry
+from auditlog_tests.models import SimpleModel
+
+
+class TwoStepMigrationTest(TestCase):
+    def test_use_text_changes_first(self):
+        text_obj = '{"field": "changes_text"}'
+        json_obj = {"field": "changes"}
+        _params = [(True, None, {"field": "changes_text"}), (False, json_obj, json_obj)]
+
+        for setting_value, changes_value, expected in _params:
+            with self.subTest():
+                entry = LogEntry(changes=changes_value, changes_text=text_obj)
+                with self.settings(
+                    AUDITLOG_USE_TEXT_CHANGES_IF_JSON_IS_NOT_PRESENT=setting_value
+                ):
+
+                    from auditlog import models
+
+                    models.changes_func = models._changes_func()
+                    self.assertEqual(entry.changes_dict, expected)
+
+
+class AuditlogMigrateJsonTest(TestCase):
+    def make_logentry(self):
+        model = SimpleModel.objects.create(text="I am a simple model.")
+        log_entry: LogEntry = model.history.first()
+        log_entry.changes_text = json.dumps(log_entry.changes)
+        log_entry.changes = None
+        log_entry.save()
+        return log_entry
+
+    def call_command(self, *args, **kwargs):
+        outbuf = StringIO()
+        errbuf = StringIO()
+        call_command(
+            "auditlogmigratejson", *args, stdout=outbuf, stderr=errbuf, **kwargs
+        )
+        return outbuf.getvalue().strip(), errbuf.getvalue().strip()
+
+    def test_using_django(self):
+        # Arrange
+        log_entry = self.make_logentry()
+
+        # Act
+        outbuf, errbuf = self.call_command()
+        log_entry.refresh_from_db()
+
+        # Assert
+        self.assertEqual(errbuf, "")
+        self.assertIsNotNone(log_entry.changes)
+
+    def test_native_postgres(self):
+        # Arrange
+        log_entry = self.make_logentry()
+
+        # Act
+        outbuf, errbuf = self.call_command("-d=postgres")
+        log_entry.refresh_from_db()
+
+        # Assert
+        self.assertEqual(errbuf, "")
+        self.assertIsNotNone(log_entry.changes)