from service.tasks.guess_system_fields import get_system_labels from .data_checker import DataChecker class FieldMapChecker(DataChecker): """ The FieldMapChecker class based on DataChecker :param data: Any variable that needs to be verified :attribute system_labels_source_df: the list of all available system fields :attribute fields_list: list of all the system fields provided in data :attribute index_list: list of all index values provided in data :attribute keys_list: list of lists that represents keys in each dict It checks for: - Data type is list - Data consists of {'value':'dummy', 'index': 0} dicts only - At least one system field is picked - No duplicate system fields - All provided system fields are valid (present in commons.system_label) - Each index has only one system field assigned to it - Check that all index values are ints """ def __init__(self, data): DataChecker.__init__(self, data) self.system_labels_source_df = get_system_labels(exclude_non_regex=False) self.fields_list = [] self.index_list = [] self.keys_list = [] def verify(self): # Checking the data type if self.d_type is not list: self.save_error("Wrong data type") else: if self._check_dict_keys(): self._check_dict_length() # Removing the fields that were not picked as system fields while "" in self.fields_list: self.fields_list.remove("") if self._check_system_fields_picked(): if self._check_duplicate_system_fields(): if self._check_if_fields_are_valid(): if self._check_duplicate_index(): self._check_index_is_int() # TODO! ADD checking if one of the suggested system fields is userEmail return self.generate_response() def _check_dict_keys(self) -> bool: """ Checking the internal dicts'structure for having 'value', 'index' pairs :return: bool """ try: for x in self.data: self.fields_list.append(x["value"]) self.index_list.append(x["index"]) self.keys_list.append(list(x.keys())) return True except Exception: self.save_error("Unexpected keys") return False def _check_dict_length(self) -> bool: """ Checking that all dicts have only 2 parameters :return: bool """ keys_set = set(len(x) for x in self.keys_list) if len(keys_set) != 1 or list(keys_set)[0] != 2: self.save_error("Incorrect number of keys") return False else: return True def _check_system_fields_picked(self) -> bool: """ Checking if any system fields were picked :return: bool """ if len(self.fields_list) == 0: self.save_error("No system fields") return False else: return True def _check_duplicate_system_fields(self) -> bool: """ Checking if there are any duplicate system fields :return: bool """ fields_set = set(self.fields_list) if len(self.fields_list) != len(fields_set): self.save_error("Duplicate fields") return False else: return True def _check_if_fields_are_valid(self) -> bool: """ Checking if all the provided system fields are valid :return: """ system_labels_source = list(self.system_labels_source_df["a_system_name"]) fields_set = set(self.fields_list) diff = fields_set.difference(system_labels_source) if len(diff) > 0: self.save_error(f"Wrong field names - {diff}") return False else: return True def _check_duplicate_index(self) -> bool: """ Checking if several fields assigned to same index :return: bool """ index_set = set(self.index_list) if len(self.index_list) != len(index_set): self.save_error("Multiple fields assigned to the same column index") return False else: return True def _check_index_is_int(self) -> bool: """ Checking if all index values are ints :return: bool """ if not all(type(x) is int for x in self.index_list): self.save_error("Wrong column index data type") return False else: return True