Coverage for dataexcept/base.py: 95%
84 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-27 14:44 +0000
1"""The root of the DataExcept exception hierarchy.
3Every operational exception this package defines derives from
4:class:`DataExceptError`, so a caller can catch the whole library with one
5clause while still catching narrowly where it matters::
7 try:
8 run_pipeline()
9 except ValidationError:
10 ... # exactly this failure
11 except DataExceptError:
12 ... # anything else DataExcept raised
14(Constructors also raise plain ``TypeError`` when given invalid arguments.
15Those are programming errors, not operational ones, and are deliberately not
16part of this hierarchy.)
18The base also carries the serialization contract for the hierarchy. Two
19problems make that necessary:
21* Most constructors take several arguments while ``Exception.args`` holds only
22 the rendered message, so the default protocol -- which replays ``args``
23 through ``__init__`` -- cannot rebuild them.
24* Several exceptions accept arbitrary caller state (``DataValidationError``
25 takes any ``value``), and that state may not be pickleable at all.
27An exception that cannot cross a process boundary is useless exactly where a
28data pipeline needs it most, so rather than fail, unpickleable state is
29replaced by a description of what was there.
30"""
32from __future__ import annotations
34import pickle
35from typing import Any, Dict, Optional, Tuple, Type
37from .failure_metadata import FailureKind, FailureMetadata
38from .redaction import redact_urls_in_text
40__all__ = ["DataExceptError", "UnpicklableCause", "UnpicklableValue"]
42_CAUSE_ATTRIBUTES = ("original", "original_exception", "cause")
45class UnpicklableValue:
46 """Stands in for state that could not survive serialization."""
48 __slots__ = ("description",)
50 def __init__(self, description: str) -> None:
51 self.description = description
53 def __repr__(self) -> str:
54 return f"<unpicklable: {self.description}>"
56 def __str__(self) -> str:
57 return self.__repr__()
59 def __eq__(self, other: object) -> bool:
60 return (
61 isinstance(other, UnpicklableValue)
62 and other.description == self.description
63 )
65 def __hash__(self) -> int:
66 return hash(self.description)
69def _safe(value: Any) -> Any:
70 """Return *value*, or a placeholder if it cannot be pickled."""
71 try:
72 pickle.dumps(value)
73 except Exception:
74 try:
75 description = f"{type(value).__name__}: {value!r}"
76 except Exception: # pragma: no cover
77 description = type(value).__name__
78 return UnpicklableValue(description[:200])
79 return value
82def _safe_exception(exc: Optional[BaseException]) -> Optional[BaseException]:
83 """Return *exc*, or an exception describing it if it cannot be pickled."""
84 if exc is None:
85 return None
86 try:
87 pickle.dumps(exc)
88 except Exception:
89 return UnpicklableCause(f"{type(exc).__name__}: {exc}")
90 return exc
93def _rebuild(
94 cls: Type["DataExceptError"],
95 args: Tuple[Any, ...],
96 state: Dict[str, Any],
97 cause: Optional[BaseException] = None,
98 context: Optional[BaseException] = None,
99 suppress_context: bool = False,
100) -> "DataExceptError":
101 """Recreate *cls* without replaying its ``__init__``."""
102 exc = cls.__new__(cls)
103 Exception.__init__(exc, *args)
104 exc.__dict__.update(state)
105 exc.__cause__ = cause
106 exc.__context__ = context
107 exc.__suppress_context__ = suppress_context
108 return exc
111class DataExceptError(Exception):
112 """Base class for every operational exception DataExcept raises."""
114 _keep_url_path = True
115 _default_failure_metadata = FailureMetadata()
117 def __init__(self, *args: Any) -> None:
118 keep_path = type(self)._keep_url_path
119 if args and isinstance(args[0], str): 119 ↛ 122line 119 didn't jump to line 122 because the condition on line 119 was always true
120 args = (redact_urls_in_text(args[0], keep_path=keep_path),) + args[1:]
122 for name, value in list(self.__dict__.items()):
123 if isinstance(value, str) and "://" in value:
124 self.__dict__[name] = redact_urls_in_text(value, keep_path=keep_path)
126 super().__init__(*args)
127 for attribute in _CAUSE_ATTRIBUTES:
128 candidate = getattr(self, attribute, None)
129 if isinstance(candidate, BaseException):
130 self.__cause__ = candidate
131 break
133 @property
134 def failure_metadata(self) -> FailureMetadata:
135 override = self.__dict__.get("_failure_metadata_override")
136 if isinstance(override, FailureMetadata):
137 return override
138 return type(self)._default_failure_metadata
140 @property
141 def failure_kind(self) -> FailureKind:
142 return self.failure_metadata.failure_kind
144 @property
145 def retryable(self) -> bool | None:
146 return self.failure_metadata.retryable
148 @property
149 def retry_after_seconds(self) -> float | None:
150 return self.failure_metadata.retry_after_seconds
152 def with_failure_metadata(self, metadata: FailureMetadata) -> "DataExceptError":
153 """Attach backend-informed metadata and return ``self`` for chaining."""
154 if not isinstance(metadata, FailureMetadata):
155 raise TypeError("metadata must be a FailureMetadata instance")
156 self._failure_metadata_override = metadata
157 return self
159 def __reduce__(self) -> Tuple[Any, Tuple[Any, ...]]:
160 args = tuple(_safe(arg) for arg in self.args)
161 state = {key: _safe(value) for key, value in self.__dict__.items()}
162 return (
163 _rebuild,
164 (
165 type(self),
166 args,
167 state,
168 _safe_exception(self.__cause__),
169 _safe_exception(self.__context__),
170 self.__suppress_context__,
171 ),
172 )
175class UnpicklableCause(DataExceptError):
176 """Stands in for a cause that could not be serialized.
178 ``__cause__`` and ``__context__`` must be exceptions, so the placeholder
179 used for ordinary attributes will not do here. Dropping the chain instead
180 would silently lose the reason for the failure.
181 """