feat: add cronjob to delete unused hardware status rows every week

Author: gustavobtflores

backend/kernelCI/settings.py

@@ -179,6 +179,13 @@ def get_json_env_var(name, default):
                 "--yes",
             ],
         ),
+        (
+            "0 0 * * 1",
+            "django.core.management.call_command",
+            [
+                "delete_unused_hardware_status",
+            ],
+        ),
     ]
 
 # Email settings for SMTP backend

backend/kernelCI_app/management/commands/delete_unused_hardware_status.py

@@ -0,0 +1,92 @@
+"""
+Management command to delete unused entries from hardware_status table.
+
+Removes HardwareStatus entries that have no corresponding checkout_id in the LatestCheckout table.
+"""
+
+import logging
+from django.core.management.base import BaseCommand
+from django.db import transaction
+from kernelCI_app.models import HardwareStatus, LatestCheckout
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    help = (
+        "Delete HardwareStatus entries with no corresponding checkout_id "
+        "in the LatestCheckout table"
+    )
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--dry-run",
+            action="store_true",
+            help="Show what would be deleted without actually deleting",
+        )
+        parser.add_argument(
+            "--batch-size",
+            type=int,
+            default=1000,
+            help="Number of records to delete per batch (default: 1000)",
+        )
+
+    def handle(self, *args, **options):
+        dry_run = options["dry_run"]
+        batch_size = options["batch_size"]
+
+        valid_checkout_ids = set(
+            LatestCheckout.objects.values_list("checkout_id", flat=True)
+        )
+
+        orphaned_entries = HardwareStatus.objects.exclude(
+            checkout_id__in=valid_checkout_ids
+        )
+        orphaned_count = orphaned_entries.count()
+
+        if orphaned_count == 0:
+            self.stdout.write(
+                self.style.SUCCESS("No orphaned HardwareStatus entries found.")
+            )
+            return
+
+        if dry_run:
+            self.stdout.write(
+                self.style.WARNING(
+                    f"[DRY RUN] Would delete {orphaned_count} entries. "
+                    "Run without --dry-run to execute deletion."
+                )
+            )
+            return
+
+        self.stdout.write(
+            f"Found {orphaned_count} HardwareStatus entries "
+            "with no corresponding LatestCheckout."
+        )
+
+        total_deleted = 0
+        while True:
+            with transaction.atomic():
+                batch_ids = list(
+                    orphaned_entries.values_list("checkout_id", flat=True)[:batch_size]
+                )
+
+                if not batch_ids:
+                    break
+
+                deleted_count = 0
+                deleted_count = HardwareStatus.objects.filter(
+                    checkout_id__in=batch_ids
+                ).delete()[0]
+
+                total_deleted += deleted_count
+                self.stdout.write(
+                    f"Deleted batch of {deleted_count} entries "
+                    f"(total: {total_deleted}/{orphaned_count})"
+                )
+
+        self.stdout.write(
+            self.style.SUCCESS(
+                f"Successfully deleted {total_deleted} orphaned HardwareStatus entries."
+            )
+        )