diff --git a/konova/management/commands/mass_delete_entries.py b/konova/management/commands/mass_delete_entries.py index 80def377..5d380d8c 100644 --- a/konova/management/commands/mass_delete_entries.py +++ b/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.") \ No newline at end of file