Merge pull request '# Mass delete script' (#576) from 573_Script_mass_delete into master

Reviewed-on: #576

konova/management/commands/mass_delete_entries.py

@@ -17,16 +17,16 @@ class Command(BaseKonovaCommand):
 
     _DELETE_CLS = None
     _DELETE_COMMENT = None
-    _DELETE_IDENTIFIER_ILIKE_PARAM = None
     _DELETE_USER = None
+    _DELETE_FROM_DB = False
 
     def handle(self, *args, **options):
         self._get_delete_user()
         self._get_object_type()
-        self._get_identifier_ilike_param()
+        filter = self._get_filter()
         self._get_delete_comment()
-        self._show_config()
-        entries = self._collect_entries()
+        self._show_config(filter)
+        entries = self._collect_entries(filter)
         self._process_delete(entries)
 
     def _get_delete_user(self):
@@ -62,36 +62,144 @@ class Command(BaseKonovaCommand):
             self._write_error(f"'{object_type_select}' not a valid option of {object_types_str}. Abort.")
             exit(-1)
 
-    def _get_identifier_ilike_param(self):
-        self._DELETE_IDENTIFIER_ILIKE_PARAM = input("Use this substring to search in identifiers: ")
+    def _get_filter(self):
+        filter_attribute = input("What do you want to filter entries by? (identifier|created_by_user): ").lower()
+        filter = {}
+        try:
+            assert filter_attribute in ["identifier", "created_by_user"]
+            if filter_attribute == "identifier":
+                filter_attribute = "identifier__contains"
+            elif filter_attribute == "created_by_user":
+                filter_attribute = "created__user__username"
+        except AssertionError:
+            self._write_error(f"'{filter_attribute}' unknown. Abort")
+            exit(-1)
+
+        filter_val = input("Use this value to search for: ")
+        try:
+            assert len(filter_val) > 0
+        except AssertionError:
+            self._write_error(f"'{filter_val}' is empty. Abort.")
+            exit(-1)
+
+        filter[filter_attribute] = filter_val
+        return filter
 
     def _get_delete_comment(self):
         self._DELETE_COMMENT = input("Comment to store in delete action: ")
 
-    def _show_config(self):
+    def _show_config(self, filter: dict):
         assert self._DELETE_CLS is not None
-        assert self._DELETE_IDENTIFIER_ILIKE_PARAM is not None
         assert self._DELETE_COMMENT is not None
 
         self._write_warning("You are about to delete entries with:")
         self._write_warning(f"    Object type: {self._DELETE_CLS.__name__}")
-        self._write_warning(f"    Identifier pattern: '*{self._DELETE_IDENTIFIER_ILIKE_PARAM}*'")
-        self._write_warning(f"    Comment for delete action: '{self._DELETE_COMMENT}'")
+        self._write_warning(f"    Filter by: '{filter}'")
         self._write_warning(f"    Delete performing user: '{self._DELETE_USER}'")
 
-    def _collect_entries(self):
+    def _collect_entries(self, filter: dict):
         return self._DELETE_CLS.objects.filter(
-            deleted__isnull=True,
-            identifier__contains=self._DELETE_IDENTIFIER_ILIKE_PARAM
+            deleted__isnull=True
+        ).filter(
+            **filter
         )
 
     def _process_delete(self, entries: QuerySet):
+        """ Wrapping delete method. Forces user to decide whether entries will be marked as deleted or shall
+        be removed completely from database
+
+        Args:
+            entries (QuerySet): Entries to be processed
+
+        Returns:
+
+        """
+        delete_type = input("Do you want 1) mark entries as deleted or 2) remove them from the database? (1|2): ")
+        try:
+            delete_type = int(delete_type)
+            assert delete_type in (1, 2)
+            self._DELETE_FROM_DB = delete_type == 2
+        except (ValueError, AssertionError) as e:
+            self._write_error(f"Entered unknown '{delete_type}'. Abort.")
+            exit(-1)
+        if self._DELETE_FROM_DB:
+            self._process_delete_from_db(entries)
+        else:
+            self._process_mark_as_deleted(entries)
+
+
+    def _process_mark_as_deleted(self, entries: QuerySet):
+        """ Method do mark chosen entries as deleted
+
+        Args:
+            entries (QuerySet): Entries to be processed
+
+        Returns:
+
+        """
+        self._write_warning(f"    Comment for delete action: '{self._DELETE_COMMENT}'")
+
+        self._check_entries_sample(entries)
+
+        entries_count = entries.count()
+        i = 0
+        for element in entries:
+            if i % 10 == 0:
+                self._write_warning(f"    Deleted {i}/{entries_count}")
+            element.mark_as_deleted(user=self._DELETE_USER, send_mail=False, comment=self._DELETE_COMMENT)
+            i += 1
+        self._write_warning(f"    Deleted {entries_count}/{entries_count}")
+        self._write_success("Entries deleted. Bye.")
+
+    def _process_delete_from_db(self, entries: QuerySet):
+        """ Method do REMOVE entries from the database
+
+        Args:
+            entries (QuerySet): Entries to be processed
+
+        Returns:
+
+        """
+        self._check_entries_sample(entries)
+
+        entries_count = entries.count()
+        input_num_entries = input(f"THIS IS THE FINAL WARNING! {entries_count} ENTRIES WILL BE REMOVED FROM THE DATABASE! ENTER THE NUMBER OF ENTRIES TO BE DELETED TO CONTINUE: ")
+        try:
+            input_num_entries = int(input_num_entries)
+        except ValueError:
+            self._write_error(f"'{input_num_entries}' is not an integer. Abort.")
+            exit(-1)
+        if entries_count == input_num_entries:
+            i = 0
+            for element in entries:
+                if i % 10 == 0:
+                    self._write_warning(f"    Deleted {i}/{entries_count}")
+                #element.delete()  # call individual delete routine to enforce custom delete logic (not called on bulk deleting!)
+                i += 1
+
+            self._write_warning(f"    Deleted {entries_count}/{entries_count}")
+            self._write_success("Entries deleted. Bye.")
+        else:
+            self._write_error(f"'{input_num_entries}' was not correct. Abort.")
+            exit(-1)
+
+    def _check_entries_sample(self, entries):
+        """ Shows a sample from the entries on the terminal, so the user can decide whether the results are as expected
+
+        Args:
+            entries (QuerySet): Entries to be processed
+
+        Returns:
+
+        """
+
         entries_count = entries.count()
         if entries_count == 0:
             self._write_success(f"Found {entries_count} matches to delete. Nothing to do here. Bye.")
             exit(0)
 
-        self._write_warning(f"Found {entries_count} matches to delete. Please double-check whether these samples from the results are what you expected: ")
+        self._write_warning(
+            f"Found {entries_count} matches to delete. Please double-check whether these samples from the results are what you expected: ")
         first_three_elements = entries[:3]
         for element in first_three_elements:
             self._write_warning(f"    {element.identifier} ({element.id})")
@@ -99,15 +207,5 @@ class Command(BaseKonovaCommand):
         delete_input = input("Are you ready to delete them now? (y|n): ").lower()
         delete_now = delete_input == "y"
         if not delete_now:
-            self._write_error(f"Entered unknown '{delete_input}'. Abort.")
+            self._write_error(f"Entered '{delete_input}'. Abort.")
             exit(-1)
-        else:
-            i = 0
-            for element in entries:
-                if i % 10 == 0:
-                    self._write_warning(f"    Deleted {i}/{entries_count}")
-                element.mark_as_deleted(user=self._DELETE_USER, send_mail=False, comment=self._DELETE_COMMENT)
-                i += 1
-
-            self._write_warning(f"    Deleted {entries_count}/{entries_count}")
-            self._write_success("Entries deleted. Bye.")