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 | """For marshmallow schemas that never fail on load, but instead just remove keys."""
from collections.abc import Mapping
from contextvars import ContextVar
from pathlib import Path
from typing import Any
from loguru import logger
from marshmallow import Schema
from marshmallow.error_store import ErrorStore, merge_errors
from typing_extensions import override
# The archive path of the file currently being loaded/dumped, used to
# prefix warning messages. A ContextVar rather than schema instance state:
# schema instances are cached process-wide (see schemas/cache.py) and
# shared across the -j N worker threads, so an instance attribute would
# race — one thread's set_path would relabel another thread's in-flight
# warnings. Each thread (and each task in an event loop) sees its own
# value.
_current_path: ContextVar[str | None] = ContextVar("comicbox_schema_path", default=None)
class ClearingErrorStore(ErrorStore):
"""Take over error processing."""
def _clean_error_list(
self,
key: str,
error_list: list[str],
cleaned_errors: dict[Any, Any],
) -> None:
if cleaned_error_list := frozenset(error_list) - self._ignore_errors:
cleaned_errors[key] = sorted(cleaned_error_list)
def _clear_errors(self) -> None:
if not self.errors:
return
cleaned_errors = {}
if isinstance(self.errors, Mapping):
for key, error_list in self.errors.items():
self._clean_error_list(key, error_list, cleaned_errors)
else:
self._clean_error_list("UNKNOWN", self.errors, cleaned_errors)
self.clear_errors = merge_errors(self.clear_errors, cleaned_errors)
self.errors = {}
def __init__(
self,
error_store: ErrorStore,
data: Mapping[str, Any],
path: str | None = None,
ignore_errors: frozenset | None = None,
) -> None:
"""Take over error processing."""
super().__init__()
self._path = path
self.clear_errors = {}
self.errors = error_store.errors
error_store.errors = {}
self._data = data
self._ignore_errors = ignore_errors or frozenset()
self._clear_errors()
@override
def store_error(self, *args: Any, **kwargs: Any) -> None:
"""Store error, but process and clear it."""
super().store_error(*args, **kwargs)
self._clear_errors()
class ClearingErrorStoreSchema(Schema):
"""Suppress Marshmallow errors to skip errored fields."""
SUPPRESS_ERRORS: bool = True
_IGNORE_ERRORS: frozenset[str] = frozenset({"Field may not be null."})
@property
def _path(self) -> str | None:
"""Current thread's file-path log prefix (see _current_path)."""
return _current_path.get()
def set_path(self, path: Path | str | None) -> None:
"""Set this thread's path prefix for error messages."""
_current_path.set(str(path) if path else None)
def __init__(
self,
path: Path | str | None = None,
ignore_errors: list | tuple | frozenset | set | None = None,
**kwargs: Any,
) -> None:
"""Initialize path and always use partial."""
if path is not None:
# Only set when explicitly supplied: marshmallow lazily
# constructs Nested sub-schemas with no path mid-load, and
# that must not clear the in-flight file's prefix.
self.set_path(path)
kwargs["partial"] = True
ignore_errors = (
frozenset() if ignore_errors is None else frozenset(ignore_errors)
)
self._ignore_errors = frozenset(ignore_errors) | self._IGNORE_ERRORS
super().__init__(**kwargs)
@override
def _deserialize(
self,
data: Any,
*,
error_store: ErrorStore,
**kwargs: Any,
) -> list | dict:
"""Skip keys and log warnings instead of throwing validation or type errors."""
if self.SUPPRESS_ERRORS:
error_store = ClearingErrorStore(
error_store, data, self._path, ignore_errors=self._ignore_errors
)
return super()._deserialize(data, error_store=error_store, **kwargs)
@override
def _invoke_field_validators(
self,
*,
error_store: ErrorStore,
data: dict[str, Any],
**kwargs: Any,
) -> None:
"""Skip keys and log warnings instead of throwing validation or type errors."""
if self.SUPPRESS_ERRORS:
error_store = ClearingErrorStore(
error_store, data, self._path, ignore_errors=self._ignore_errors
)
super()._invoke_field_validators(error_store=error_store, data=data, **kwargs)
@override
def _invoke_schema_validators(
self,
*,
error_store: ErrorStore,
data,
**kwargs,
) -> None:
"""Skip keys and log warnings instead of throwing validation or type errors."""
if self.SUPPRESS_ERRORS:
error_store = ClearingErrorStore(
error_store, data, self._path, ignore_errors=self._ignore_errors
)
super()._invoke_schema_validators(error_store=error_store, **kwargs)
def _filter_list(self, error_list: list) -> list:
return sorted(frozenset(error_list) - self._ignore_errors)
def _filter_mapping(self, error: Mapping) -> dict:
return {
key: filtered
for key, error_list in error.items()
if (filtered := self._filter_list(error_list))
}
def _log_warnings(
self,
error_class: type | None,
errors: Mapping | list,
) -> None:
if not errors:
return
path = f"{self._path}: " if self._path else ""
error_name = f"{error_class.__name__} - " if error_class else ""
message = f"{path}{error_name}{errors}"
logger.warning(message)
@override
def handle_error(
self,
error: Any,
*_args: Any,
**_kwargs: Any,
) -> None:
"""Log unignored errors at WARNING; ignored errors are dropped."""
if hasattr(error, "normalized_messages"):
error_class = type(error)
error = error.normalized_messages()
elif hasattr(error, "message"):
error_class = type(error)
error = error.message
else:
error_class = None
if isinstance(error, Mapping):
warning_errors = self._filter_mapping(error)
else:
error_list = error if isinstance(error, list) else [error]
warning_errors = self._filter_list(error_list)
self._log_warnings(error_class, warning_errors)
|