add task and management command for cleaning up logs

log_outgoing_requests/conf.py

@@ -52,6 +52,12 @@ class LogOutgoingRequestsConf(AppConf):
     database, but the body will be missing.
     """
 
+    MAX_AGE = None
+    """
+    The maximum age of request logs, after which they are deleted (via a Celery task,
+    Django management command, or the like).
+    """
+
     RESET_DB_SAVE_AFTER = 60
     """
     If the config has been updated, reset the database logging after the specified

log_outgoing_requests/management/commands/cleanup_requests_logs.py

@@ -0,0 +1,8 @@
+from django.core.management.base import BaseCommand
+
+from log_outgoing_requests.models import OutgoingRequestsLog
+
+
+class Command(BaseCommand):
+    def handle(self, *args, **kwargs):
+        OutgoingRequestsLog.objects.cleanup_requests_logs()

log_outgoing_requests/models.py

@@ -1,9 +1,11 @@
 import logging
+from datetime import timedelta
 from typing import Union
 from urllib.parse import urlparse
 
 from django.core.validators import MinValueValidator
 from django.db import models
+from django.utils import timezone
 from django.utils.functional import cached_property
 from django.utils.translation import gettext_lazy as _
 
@@ -16,6 +18,18 @@
 logger = logging.getLogger(__name__)
 
 
+class OutgoingRequestsLogQueryset(models.QuerySet):
+    def cleanup_requests_logs(self):
+        max_age = settings.LOG_OUTGOING_REQUESTS_MAX_AGE
+        if max_age is None:
+            return
+
+        now = timezone.now()
+        num_deleted, _ = self.filter(timestamp__lt=now - timedelta(max_age)).delete()
+        logger.info("Deleted %d outgoing request log(s)", num_deleted)
+        return num_deleted
+
+
 class OutgoingRequestsLog(models.Model):
     url = models.URLField(
         verbose_name=_("URL"),
@@ -108,6 +122,8 @@ class OutgoingRequestsLog(models.Model):
         help_text=_("Text providing information in case of request failure."),
     )
 
+    objects = OutgoingRequestsLogQueryset.as_manager()
+
     class Meta:
         verbose_name = _("Outgoing request log")
         verbose_name_plural = _("Outgoing request logs")

log_outgoing_requests/tasks.py

@@ -2,6 +2,13 @@
 from .constants import SaveLogsChoice
 
 
+@shared_task
+def cleanup_requests_logs():
+    from .models import OutgoingRequestsLog
+
+    OutgoingRequestsLog.objects.cleanup_requests_logs()
+
+
 @shared_task
 def reset_config():
     from .models import OutgoingRequestsLogConfig

tests/test_tasks.py

@@ -1,5 +1,9 @@
+from django.core.management import call_command
+from django.utils import timezone
+
 import pytest
 import requests
+from freezegun import freeze_time
 
 from log_outgoing_requests.config_reset import schedule_config_reset
 from log_outgoing_requests.models import OutgoingRequestsLog, OutgoingRequestsLogConfig
@@ -12,6 +16,35 @@
 has_celery = celery is not None
 
 
+@pytest.mark.django_db
+def test_cleanup_request_logs_command(settings):
+    settings.LOG_OUTGOING_REQUESTS_MAX_AGE = 1  # delete if > 1 day old
+
+    with freeze_time("2023-10-02T12:00:00Z") as frozen_time:
+        OutgoingRequestsLog.objects.create(timestamp=timezone.now())
+        frozen_time.move_to("2023-10-04T12:00:00Z")
+        recent_log = OutgoingRequestsLog.objects.create(timestamp=timezone.now())
+        call_command("cleanup_requests_logs")
+
+    assert OutgoingRequestsLog.objects.get() == recent_log
+
+
+@pytest.mark.skipif(not has_celery, reason="Celery is optional dependency")
+@pytest.mark.django_db
+def test_cleanup_request_logs_celery_task(requests_mock, settings):
+    from log_outgoing_requests.tasks import cleanup_requests_logs
+
+    settings.LOG_OUTGOING_REQUESTS_MAX_AGE = 1  # delete if > 1 old old
+
+    with freeze_time("2023-10-02T12:00:00Z") as frozen_time:
+        OutgoingRequestsLog.objects.create(timestamp=timezone.now())
+        frozen_time.move_to("2023-10-04T12:00:00Z")
+        recent_log = OutgoingRequestsLog.objects.create(timestamp=timezone.now())
+        cleanup_requests_logs()
+
+    assert OutgoingRequestsLog.objects.get() == recent_log
+
+
 @pytest.mark.skipif(not has_celery, reason="Celery is optional dependency")
 @pytest.mark.django_db
 def test_reset_config(requests_mock, settings):