#35 Sanity command

* adds new sanitize_db.py command (WIP)

konova/management/commands/sanitize_db.py

@@ -0,0 +1,108 @@
+"""
+Author: Michel Peltriaux
+Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
+Contact: michel.peltriaux@sgdnord.rlp.de
+Created on: 16.11.21
+
+"""
+from django.db.models import QuerySet
+
+from compensation.models import CompensationState, Compensation, EcoAccount, AbstractCompensation
+from ema.models import Ema
+from konova.management.commands.setup import BaseKonovaCommand
+from user.models import UserActionLogEntry
+
+
+class Command(BaseKonovaCommand):
+    help = "Checks the database' sanity and removes unused entries"
+
+    def handle(self, *args, **options):
+        try:
+            #self.sanitize_log_entries()
+            self.sanitize_compensation_states()
+        except KeyboardInterrupt:
+            self._break_line()
+            exit(-1)
+
+    def sanitize_log_entries(self):
+        """ Removes log entries which are not attached to any logs
+
+        Returns:
+
+        """
+        all_log_entries = UserActionLogEntry.objects.all()
+        delete_ids = []
+        for entry in all_log_entries:
+            logs_intervention = entry.intervention_set.all()
+            logs_compensation = entry.compensation_set.all()
+            logs_account = entry.ecoaccount_set.all()
+            logs_ema = entry.ema_set.all()
+            logs_something = logs_intervention or logs_compensation or logs_account or logs_ema
+            if not logs_something:
+                delete_ids.append(entry.id)
+        entries_to_delete = UserActionLogEntry.objects.filter(
+            id__in=delete_ids
+        )
+        num_entries = entries_to_delete.count()
+        if num_entries > 0:
+            self._write_warning(f"Found {num_entries} log entries not attached to anything. Delete now...")
+            entries_to_delete.delete()
+            self._write_success("Log entries deleted.")
+        else:
+            self._write_success("No unattached log entries found. Everything clear.")
+
+    def get_all_state_ids(self, cls):
+        """ Getter for all states (before and after) of a class
+
+        Args:
+            cls ():
+
+        Returns:
+
+        """
+        if not isinstance(cls, AbstractCompensation):
+            return QuerySet.none()
+
+        all_objects = cls.objects.all().prefetch_related(
+            "before_states",
+            "after_states",
+        )
+        before_state_ids = all_objects.exclude(
+            before_states__isnull=True,
+        ).values_list(
+            "before_states__id",
+            flat=True,
+        )
+        after_state_ids = all_objects.exclude(
+            after_states__isnull=True,
+        ).values_list(
+            "after_states__id",
+            flat=True,
+        )
+        all_ids = after_state_ids.union(before_state_ids)
+        return all_ids
+
+    def sanitize_compensation_states(self):
+        """ Removes unattached compensation states
+
+        Returns:
+
+        """
+        all_states = CompensationState.objects.all()
+        compensation_state_ids = self.get_all_state_ids(Compensation)
+        account_state_ids = self.get_all_state_ids(EcoAccount)
+        ema_state_ids = self.get_all_state_ids(Ema)
+        all_state_ids = compensation_state_ids.union(account_state_ids, ema_state_ids)
+
+        all_states = all_states.exclude(
+            id__in=all_state_ids
+        )
+        num_all_states = all_states.count()
+        if num_all_states > 0:
+            self._write_warning(f"Found {num_all_states} unused compensation states. Delete now...")
+            all_states.delete()
+            self._write_success("Unused states deleted.")
+        else:
+            self._write_success("No unused states found!")
+
+