-
-
Notifications
You must be signed in to change notification settings - Fork 797
/
results.py
265 lines (215 loc) · 8.49 KB
/
results.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
import logging
import sys
import traceback
from collections import OrderedDict
from django.core.exceptions import NON_FIELD_ERRORS
from django.utils.encoding import force_str
from django.utils.functional import cached_property
from django.utils.translation import gettext_lazy as _
from tablib import Dataset
logger = logging.getLogger(__name__)
class Error:
def __init__(self, error, row=None, number=None):
self.error = error
self.row = row
self.number = number
def __repr__(self):
result = f"<Error: {self.error!r}"
if self.row is not None:
result += f" at row {self.row}"
if self.number is not None:
result += f" at number {self.number}"
result += ">"
return result
@cached_property
def traceback(self):
if sys.version_info >= (3, 10):
lines = traceback.format_exception(self.error)
else:
lines = traceback.format_exception(
None, self.error, self.error.__traceback__
)
return "".join(lines)
class RowResult:
"""Container for values relating to a row import."""
IMPORT_TYPE_UPDATE = "update"
IMPORT_TYPE_NEW = "new"
IMPORT_TYPE_DELETE = "delete"
IMPORT_TYPE_SKIP = "skip"
IMPORT_TYPE_ERROR = "error"
IMPORT_TYPE_INVALID = "invalid"
valid_import_types = frozenset(
[
IMPORT_TYPE_NEW,
IMPORT_TYPE_UPDATE,
IMPORT_TYPE_DELETE,
IMPORT_TYPE_SKIP,
]
)
def __init__(self):
#: An instance of :class:`~import_export.results.Error` which may have been
#: raised during import.
self.errors = []
#: Contains any ValidationErrors which may have been raised during import.
self.validation_error = None
#: A HTML representation of the difference between the 'original' and
#: 'updated' model instance.
self.diff = None
#: A string identifier which identifies what type of import was performed.
self.import_type = None
#: Retain the raw values associated with each imported row.
self.row_values = {}
#: The instance id (used in Admin UI)
self.object_id = None
#: The object representation (used in Admin UI)
self.object_repr = None
#: A reference to the model instance which was created, updated or deleted.
self.instance = None
#: A reference to the model instance before updates were applied.
#: This value is only set for updates.
self.original = None
def is_update(self):
"""
:return: ``True`` if import type is 'update', otherwise ``False``.
"""
return self.import_type == RowResult.IMPORT_TYPE_UPDATE
def is_new(self):
"""
:return: ``True`` if import type is 'new', otherwise ``False``.
"""
return self.import_type == RowResult.IMPORT_TYPE_NEW
def is_delete(self):
"""
:return: ``True`` if import type is 'delete', otherwise ``False``.
"""
return self.import_type == RowResult.IMPORT_TYPE_DELETE
def is_skip(self):
"""
:return: ``True`` if import type is 'skip', otherwise ``False``.
"""
return self.import_type == RowResult.IMPORT_TYPE_SKIP
def is_error(self):
"""
:return: ``True`` if import type is 'error', otherwise ``False``.
"""
return self.import_type == RowResult.IMPORT_TYPE_ERROR
def is_invalid(self):
"""
:return: ``True`` if import type is 'invalid', otherwise ``False``.
"""
return self.import_type == RowResult.IMPORT_TYPE_INVALID
def is_valid(self):
"""
:return: ``True`` if import type is not 'error' or 'invalid', otherwise
``False``.
"""
return self.import_type in self.valid_import_types
def add_instance_info(self, instance):
if instance is not None:
# Add object info to RowResult (e.g. for LogEntry)
self.object_id = getattr(instance, "pk", None)
try:
self.object_repr = force_str(instance)
except Exception as e:
logger.debug(_("call to force_str() on instance failed: %s" % str(e)))
class InvalidRow:
"""A row that resulted in one or more ``ValidationError``
being raised during import."""
def __init__(self, number, validation_error, values):
self.number = number
self.error = validation_error
self.values = values
try:
self.error_dict = validation_error.message_dict
except AttributeError:
self.error_dict = {NON_FIELD_ERRORS: validation_error.messages}
@property
def field_specific_errors(self):
"""Returns a dictionary of field-specific validation errors for this row."""
return {
key: value
for key, value in self.error_dict.items()
if key != NON_FIELD_ERRORS
}
@property
def non_field_specific_errors(self):
"""Returns a list of non field-specific validation errors for this row."""
return self.error_dict.get(NON_FIELD_ERRORS, [])
@property
def error_count(self):
"""Returns the total number of validation errors for this row."""
count = 0
for error_list in self.error_dict.values():
count += len(error_list)
return count
class ErrorRow:
"""A row that resulted in one or more errors being raised during import."""
def __init__(self, number, errors):
#: The row number
self.number = number
#: A list of errors associated with the row
self.errors = errors
class Result:
def __init__(self, *args, **kwargs):
super().__init__()
self.base_errors = []
self.diff_headers = []
#: The rows associated with the result.
self.rows = []
#: The collection of rows which had validation errors.
self.invalid_rows = []
#: The collection of rows which had generic errors.
self.error_rows = []
#: A custom Dataset containing only failed rows and associated errors.
self.failed_dataset = Dataset()
self.totals = OrderedDict(
[
(RowResult.IMPORT_TYPE_NEW, 0),
(RowResult.IMPORT_TYPE_UPDATE, 0),
(RowResult.IMPORT_TYPE_DELETE, 0),
(RowResult.IMPORT_TYPE_SKIP, 0),
(RowResult.IMPORT_TYPE_ERROR, 0),
(RowResult.IMPORT_TYPE_INVALID, 0),
]
)
self.total_rows = 0
def valid_rows(self):
return [r for r in self.rows if r.import_type in RowResult.valid_import_types]
def append_row_result(self, row_result):
self.rows.append(row_result)
def append_base_error(self, error):
self.base_errors.append(error)
def add_dataset_headers(self, headers):
headers = [] if not headers else headers
self.failed_dataset.headers = headers + ["Error"]
def append_failed_row(self, row, error):
row_values = [v for (k, v) in row.items()]
try:
row_values.append(str(error.error))
except AttributeError:
row_values.append(str(error))
self.failed_dataset.append(row_values)
def append_invalid_row(self, number, row, validation_error):
# NOTE: value order must match diff_headers order, so that row
# values and column headers match in the UI when displayed
values = tuple(row.get(col, "---") for col in self.diff_headers)
self.invalid_rows.append(
InvalidRow(number=number, validation_error=validation_error, values=values)
)
def append_error_row(self, number, row, errors):
self.error_rows.append(ErrorRow(number=number, errors=errors))
def increment_row_result_total(self, row_result):
if row_result.import_type:
self.totals[row_result.import_type] += 1
def row_errors(self):
return [(i + 1, row.errors) for i, row in enumerate(self.rows) if row.errors]
def has_errors(self):
"""Returns a boolean indicating whether the import process resulted in
any critical (non-validation) errors for this result."""
return bool(self.base_errors or self.row_errors())
def has_validation_errors(self):
"""Returns a boolean indicating whether the import process resulted in
any validation errors for this result."""
return bool(self.invalid_rows)
def __iter__(self):
return iter(self.rows)