Add delete historical management command

eventstore/management/commands/delete_historical_records.py

@@ -0,0 +1,38 @@
+import logging
+
+from dateutil.relativedelta import relativedelta
+from django.apps import apps
+from django.core.management.base import BaseCommand
+from django.utils import timezone
+
+logger = logging.getLogger(__name__)
+
+VALID_MODELS = {"Event", "Message"}
+
+
+class Command(BaseCommand):
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "model",
+            type=str,
+            help="Specify the model you want to delete the records for.(Event/Message)",
+        )
+        parser.add_argument(
+            "retention_period",
+            type=int,
+            help="Specify the retention period in months",
+            default=60,
+        )
+
+    def handle(self, *args, **options):
+        model_name = options["model"]
+        retention_period = options["retention_period"]
+
+        if model_name not in VALID_MODELS:
+            raise Exception("Invalid model specified")
+
+        model = apps.get_model("eventstore", model_name)
+
+        filter_date = timezone.now() - relativedelta(months=retention_period, hour=0)
+        count, _ = model.objects.filter(timestamp__lt=filter_date).delete()
+        logger.info(f"Deleted {count} {model_name.lower()}(s)")

eventstore/management/commands/tests/test_delete_historical_records.py

@@ -0,0 +1,55 @@
+from io import StringIO
+
+from dateutil.relativedelta import relativedelta
+from django.core.management import call_command
+from django.core.management.base import CommandError
+from django.test import TestCase
+from django.utils import timezone
+
+from eventstore.models import Event, Message
+
+
+class DeleteHistoricalRecordsTests(TestCase):
+    def call_command(self, *args, **kwargs):
+        out = StringIO()
+        call_command(
+            "delete_historical_records",
+            *args,
+            stdout=out,
+            stderr=StringIO(),
+            **kwargs,
+        )
+        return out.getvalue()
+
+    def create_record(self, model, id, timestamp):
+        record = model.objects.create(id=id)
+        record.timestamp = timestamp
+        record.save()
+
+    def test_missing_arguments(self):
+        self.assertRaises(CommandError, self.call_command)
+
+    def test_invalid_arguments(self):
+        self.assertRaises(Exception, self.call_command, "InvalidModel")
+
+    def test_delete_events(self):
+        running_month = timezone.now() - relativedelta(months=12, hour=12)
+
+        for i in range(12):
+            self.create_record(Event, i, running_month)
+            running_month = running_month + relativedelta(months=1)
+
+        self.call_command("Event", 6)
+
+        self.assertEqual(Event.objects.count(), 6)
+
+    def test_delete_messages(self):
+        running_month = timezone.now() - relativedelta(months=12, hour=12)
+
+        for i in range(12):
+            self.create_record(Message, i, running_month)
+            running_month = running_month + relativedelta(months=1)
+
+        self.call_command("Message", 6)
+
+        self.assertEqual(Message.objects.count(), 6)