Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/import_export/results.py: 39%

1from collections import OrderedDict 

2 

3from django.core.exceptions import NON_FIELD_ERRORS 

4from django.utils.encoding import force_str 

5from tablib import Dataset 

6 

7 

8class Error: 

9 def __init__(self, error, traceback=None, row=None): 

10 self.error = error 

11 self.traceback = traceback 

12 self.row = row 

13 

14 

15class RowResult: 

16 IMPORT_TYPE_UPDATE = 'update' 

17 IMPORT_TYPE_NEW = 'new' 

18 IMPORT_TYPE_DELETE = 'delete' 

19 IMPORT_TYPE_SKIP = 'skip' 

20 IMPORT_TYPE_ERROR = 'error' 

21 IMPORT_TYPE_INVALID = 'invalid' 

22 

23 valid_import_types = frozenset([ 

24 IMPORT_TYPE_NEW, 

25 IMPORT_TYPE_UPDATE, 

26 IMPORT_TYPE_DELETE, 

27 IMPORT_TYPE_SKIP, 

28 ]) 

29 

30 def __init__(self): 

31 self.errors = [] 

32 self.validation_error = None 

33 self.diff = None 

34 self.import_type = None 

35 self.row_values = {} 

36 self.object_id = None 

37 self.object_repr = None 

38 

39 def add_instance_info(self, instance): 

40 if instance is not None: 

41 # Add object info to RowResult (e.g. for LogEntry) 

42 self.object_id = getattr(instance, "pk", None) 

43 self.object_repr = force_str(instance) 

44 

45 

46class InvalidRow: 

47 """A row that resulted in one or more ``ValidationError`` being raised during import.""" 

48 

49 def __init__(self, number, validation_error, values): 

50 self.number = number 

51 self.error = validation_error 

52 self.values = values 

53 try: 

54 self.error_dict = validation_error.message_dict 

55 except AttributeError: 

56 self.error_dict = {NON_FIELD_ERRORS: validation_error.messages} 

57 

58 @property 

59 def field_specific_errors(self): 

60 """Returns a dictionary of field-specific validation errors for this row.""" 

61 return { 

62 key: value for key, value in self.error_dict.items() 

63 if key != NON_FIELD_ERRORS 

64 } 

65 

66 @property 

67 def non_field_specific_errors(self): 

68 """Returns a list of non field-specific validation errors for this row.""" 

69 return self.error_dict.get(NON_FIELD_ERRORS, []) 

70 

71 @property 

72 def error_count(self): 

73 """Returns the total number of validation errors for this row.""" 

74 count = 0 

75 for error_list in self.error_dict.values(): 

76 count += len(error_list) 

77 return count 

78 

79 

80class Result: 

81 def __init__(self, *args, **kwargs): 

82 super().__init__() 

83 self.base_errors = [] 

84 self.diff_headers = [] 

85 self.rows = [] # RowResults 

86 self.invalid_rows = [] # InvalidRow 

87 self.failed_dataset = Dataset() 

88 self.totals = OrderedDict([(RowResult.IMPORT_TYPE_NEW, 0), 

89 (RowResult.IMPORT_TYPE_UPDATE, 0), 

90 (RowResult.IMPORT_TYPE_DELETE, 0), 

91 (RowResult.IMPORT_TYPE_SKIP, 0), 

92 (RowResult.IMPORT_TYPE_ERROR, 0), 

93 (RowResult.IMPORT_TYPE_INVALID, 0)]) 

94 self.total_rows = 0 

95 

96 def valid_rows(self): 

97 return [ 

98 r for r in self.rows 

99 if r.import_type in RowResult.valid_import_types 

100 ] 

101 

102 def append_row_result(self, row_result): 

103 self.rows.append(row_result) 

104 

105 def append_base_error(self, error): 

106 self.base_errors.append(error) 

107 

108 def add_dataset_headers(self, headers): 

109 headers = list() if not headers else headers 

110 self.failed_dataset.headers = headers + ["Error"] 

111 

112 def append_failed_row(self, row, error): 

113 row_values = [v for (k, v) in row.items()] 

114 try: 

115 row_values.append(str(error.error)) 

116 except AttributeError: 

117 row_values.append(str(error)) 

118 self.failed_dataset.append(row_values) 

119 

120 def append_invalid_row(self, number, row, validation_error): 

121 # NOTE: value order must match diff_headers order, so that row 

122 # values and column headers match in the UI when displayed 

123 values = tuple(row.get(col, "---") for col in self.diff_headers) 

124 self.invalid_rows.append( 

125 InvalidRow(number=number, validation_error=validation_error, values=values) 

126 ) 

127 

128 def increment_row_result_total(self, row_result): 

129 if row_result.import_type: 

130 self.totals[row_result.import_type] += 1 

131 

132 def row_errors(self): 

133 return [(i + 1, row.errors) 

134 for i, row in enumerate(self.rows) if row.errors] 

135 

136 def has_errors(self): 

137 """Returns a boolean indicating whether the import process resulted in 

138 any critical (non-validation) errors for this result.""" 

139 return bool(self.base_errors or self.row_errors()) 

140 

141 def has_validation_errors(self): 

142 """Returns a boolean indicating whether the import process resulted in 

143 any validation errors for this result.""" 

144 return bool(self.invalid_rows) 

145 

146 def __iter__(self): 

147 return iter(self.rows)