Coverage for dataexcept/serialization.py: 91%
147 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-27 14:44 +0000
1"""Strict JSON-safe structured representations of exceptions."""
3from __future__ import annotations
5import builtins
6import json
7import math
8from collections.abc import Mapping, Sequence, Set
9from typing import Any
11from .base import DataExceptError
12from .redaction import redact_urls_in_text
14__all__ = ["exception_to_dict", "exception_to_json"]
16_MAX_VALUE_DEPTH = 8
17_NOT_SCALAR = object()
18_EXCEPTION_GROUP_TYPE = getattr(builtins, "BaseExceptionGroup", None)
19_UNKNOWN_FAILURE = {
20 "kind": "unknown",
21 "retryable": None,
22 "retry_after_seconds": None,
23}
26def _redact_export_text(text: str) -> str:
27 """Scrub URLs for export, including their paths."""
28 return redact_urls_in_text(text, keep_path=False)
31def _safe_text(value: Any) -> str:
32 """Render *value* without raising and scrub credential-bearing URLs."""
33 try:
34 text = str(value)
35 except Exception:
36 try:
37 text = f"<unrepresentable {type(value).__name__}>"
38 except Exception: # pragma: no cover - hostile type metadata
39 text = "<unrepresentable>"
40 return _redact_export_text(text)
43def _safe_key(value: Any) -> str:
44 if isinstance(value, str): 44 ↛ 46line 44 didn't jump to line 46 because the condition on line 44 was always true
45 return _redact_export_text(value)
46 return _safe_text(value)
49def _safe_scalar(value: Any) -> Any:
50 if value is None or isinstance(value, (bool, int)):
51 return value
52 if isinstance(value, float):
53 return value if math.isfinite(value) else str(value)
54 if isinstance(value, str):
55 return _redact_export_text(value)
56 if isinstance(value, (bytes, bytearray)):
57 return _safe_text(value)
58 return _NOT_SCALAR
61def _safe_mapping(value: Mapping[Any, Any], *, depth: int, seen: set[int]) -> Any:
62 identity = id(value)
63 seen.add(identity)
64 try:
65 return {
66 _safe_key(key): _json_safe(item, depth=depth + 1, seen=seen)
67 for key, item in value.items()
68 }
69 except Exception:
70 return _safe_text(value)
71 finally:
72 seen.discard(identity)
75def _safe_collection(
76 value: Sequence[Any] | Set[Any], *, depth: int, seen: set[int]
77) -> Any:
78 identity = id(value)
79 seen.add(identity)
80 try:
81 return [_json_safe(item, depth=depth + 1, seen=seen) for item in value]
82 except Exception:
83 return _safe_text(value)
84 finally:
85 seen.discard(identity)
88def _json_safe(value: Any, *, depth: int = 0, seen: set[int] | None = None) -> Any:
89 """Return *value* in a strict JSON-safe form without raising."""
90 if seen is None:
91 seen = set()
92 scalar = _safe_scalar(value)
93 if scalar is not _NOT_SCALAR:
94 return scalar
95 if depth >= _MAX_VALUE_DEPTH: 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true
96 return "<truncated>"
97 if id(value) in seen: 97 ↛ 98line 97 didn't jump to line 98 because the condition on line 97 was never true
98 return "<cycle>"
99 if isinstance(value, Mapping):
100 return _safe_mapping(value, depth=depth, seen=seen)
101 if isinstance(value, (Sequence, Set)):
102 return _safe_collection(value, depth=depth, seen=seen)
103 return _safe_text(value)
106def _attributes(exc: BaseException) -> dict[str, Any]:
107 """Return public instance attributes in a JSON-safe representation."""
108 try:
109 state = vars(exc)
110 result: dict[str, Any] = {}
111 for name, value in state.items():
112 if not isinstance(name, str) or name.startswith("_"):
113 continue
114 result[name] = _json_safe(value)
115 return result
116 except Exception:
117 return {}
120def _group_members(exc: BaseException) -> Sequence[BaseException] | None:
121 """Return exception-group members without importing a 3.11-only symbol."""
122 if _EXCEPTION_GROUP_TYPE is None or not isinstance(exc, _EXCEPTION_GROUP_TYPE):
123 return None
124 try:
125 members: object = getattr(exc, "exceptions", None)
126 if not isinstance(members, tuple): 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true
127 return None
128 if not all(isinstance(member, BaseException) for member in members): 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true
129 return None
130 return members
131 except Exception:
132 return None
135def _failure_record(exc: DataExceptError) -> dict[str, Any]:
136 """Return failure metadata without letting hostile overrides escape."""
137 try:
138 metadata = exc.failure_metadata
139 failure = {
140 "kind": metadata.failure_kind,
141 "retryable": metadata.retryable,
142 "retry_after_seconds": metadata.retry_after_seconds,
143 }
144 if failure["kind"] not in {"transient", "permanent", "unknown"}: 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true
145 raise ValueError("invalid failure kind")
146 if failure["retryable"] is not None and not isinstance( 146 ↛ 149line 146 didn't jump to line 149 because the condition on line 146 was never true
147 failure["retryable"], bool
148 ):
149 raise TypeError("invalid retryable value")
151 retry_after = failure["retry_after_seconds"]
152 if retry_after is not None:
153 if isinstance(retry_after, bool) or not isinstance(
154 retry_after, (int, float)
155 ):
156 raise TypeError("invalid retry_after_seconds value")
157 retry_after = float(retry_after)
158 if not math.isfinite(retry_after) or retry_after < 0:
159 raise ValueError("invalid retry_after_seconds value")
160 failure["retry_after_seconds"] = retry_after
162 json.dumps(failure, allow_nan=False)
163 return failure
164 except Exception:
165 return dict(_UNKNOWN_FAILURE)
168def _failure_fields(exc: BaseException) -> dict[str, Any]:
169 """Return the optional structured failure field for *exc*."""
170 if not isinstance(exc, DataExceptError):
171 return {}
172 return {"failure": _failure_record(exc)}
175def _exception_record(
176 exc: BaseException,
177 *,
178 include_attributes: bool,
179 max_depth: int,
180 depth: int,
181 seen: set[int],
182) -> dict[str, Any]:
183 if depth > max_depth:
184 return {"truncated": True}
186 identity = id(exc)
187 if identity in seen:
188 return {
189 "type": type(exc).__name__,
190 "module": type(exc).__module__,
191 "message": _safe_text(exc),
192 "cycle": True,
193 }
195 seen.add(identity)
196 record: dict[str, Any] = {
197 "type": type(exc).__name__,
198 "module": type(exc).__module__,
199 "message": _safe_text(exc),
200 }
201 record.update(_failure_fields(exc))
202 if include_attributes:
203 attributes = _attributes(exc)
204 if attributes:
205 record["attributes"] = attributes
207 members = _group_members(exc)
208 if members is not None:
209 record["exceptions"] = [
210 _exception_record(
211 member,
212 include_attributes=include_attributes,
213 max_depth=max_depth,
214 depth=depth + 1,
215 seen=seen,
216 )
217 for member in members
218 ]
220 if exc.__cause__ is not None:
221 record["cause"] = _exception_record(
222 exc.__cause__,
223 include_attributes=include_attributes,
224 max_depth=max_depth,
225 depth=depth + 1,
226 seen=seen,
227 )
228 if exc.__context__ is not None and not exc.__suppress_context__:
229 record["context"] = _exception_record(
230 exc.__context__,
231 include_attributes=include_attributes,
232 max_depth=max_depth,
233 depth=depth + 1,
234 seen=seen,
235 )
237 seen.discard(identity)
238 return record
241def exception_to_dict(
242 exc: BaseException,
243 *,
244 include_attributes: bool = True,
245 max_depth: int = 8,
246) -> dict[str, Any]:
247 """Return a strict JSON-safe structured representation of *exc*."""
248 if not isinstance(exc, BaseException):
249 raise TypeError("exc must be an exception instance")
250 if not isinstance(max_depth, int) or isinstance(max_depth, bool):
251 raise TypeError("max_depth must be an integer")
252 if max_depth < 0:
253 raise ValueError("max_depth must be non-negative")
254 return _exception_record(
255 exc,
256 include_attributes=include_attributes,
257 max_depth=max_depth,
258 depth=0,
259 seen=set(),
260 )
263def exception_to_json(
264 exc: BaseException,
265 *,
266 include_attributes: bool = True,
267 max_depth: int = 8,
268 **json_kwargs: Any,
269) -> str:
270 """Return :func:`exception_to_dict` encoded as strict JSON."""
271 json_kwargs["allow_nan"] = False
272 return json.dumps(
273 exception_to_dict(
274 exc,
275 include_attributes=include_attributes,
276 max_depth=max_depth,
277 ),
278 **json_kwargs,
279 )