Enhancement: prune audit logs and management command (#8416)

src/documents/management/commands/prune_audit_logs.py

@@ -0,0 +1,39 @@
+from auditlog.models import LogEntry
+from django.core.management.base import BaseCommand
+from django.db import transaction
+from tqdm import tqdm
+
+from documents.management.commands.mixins import ProgressBarMixin
+
+
+class Command(BaseCommand, ProgressBarMixin):
+    """
+    Prune the audit logs of objects that no longer exist.
+    """
+
+    help = "Prunes the audit logs of objects that no longer exist."
+
+    def add_arguments(self, parser):
+        self.add_argument_progress_bar_mixin(parser)
+
+    def handle(self, **options):
+        self.handle_progress_bar_mixin(**options)
+        with transaction.atomic():
+            for log_entry in tqdm(LogEntry.objects.all(), disable=self.no_progress_bar):
+                model_class = log_entry.content_type.model_class()
+                # use global_objects for SoftDeleteModel
+                objects = (
+                    model_class.global_objects
+                    if hasattr(model_class, "global_objects")
+                    else model_class.objects
+                )
+                if (
+                    log_entry.object_id
+                    and not objects.filter(pk=log_entry.object_id).exists()
+                ):
+                    log_entry.delete()
+                    tqdm.write(
+                        self.style.NOTICE(
+                            f"Deleted audit log entry for {model_class.__name__} #{log_entry.object_id}",
+                        ),
+                    )

src/documents/tasks.py

@@ -10,6 +10,7 @@ import tqdm
 from celery import Task
 from celery import shared_task
 from django.conf import settings
+from django.contrib.contenttypes.models import ContentType
 from django.db import models
 from django.db import transaction
 from django.db.models.signals import post_save
@@ -332,9 +333,17 @@ def empty_trash(doc_ids=None):
     )
 
     try:
+        deleted_document_ids = documents.values_list("id", flat=True)
         # Temporarily connect the cleanup handler
         models.signals.post_delete.connect(cleanup_document_deletion, sender=Document)
         documents.delete()  # this is effectively a hard delete
+
+        if settings.AUDIT_LOG_ENABLED:
+            # Delete the audit log entries for documents that dont exist anymore
+            LogEntry.objects.filter(
+                content_type=ContentType.objects.get_for_model(Document),
+                object_id__in=deleted_document_ids,
+            ).delete()
     except Exception as e:  # pragma: no cover
         logger.exception(f"Error while emptying trash: {e}")
     finally:

src/documents/tests/test_management.py

@@ -7,6 +7,8 @@ from io import StringIO
 from pathlib import Path
 from unittest import mock
 
+from auditlog.models import LogEntry
+from django.contrib.contenttypes.models import ContentType
 from django.core.management import call_command
 from django.test import TestCase
 from django.test import override_settings
@@ -252,3 +254,15 @@ class TestConvertMariaDBUUID(TestCase):
         m.assert_called_once()
 
         self.assertIn("Successfully converted", stdout.getvalue())
+
+
+class TestPruneAuditLogs(TestCase):
+    def test_prune_audit_logs(self):
+        LogEntry.objects.create(
+            content_type=ContentType.objects.get_for_model(Document),
+            object_id=1,
+            action=LogEntry.Action.CREATE,
+        )
+        call_command("prune_audit_logs")
+
+        self.assertEqual(LogEntry.objects.count(), 0)