Coverage for dataexcept/serialization.py: 85%
120 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-03 20:46 +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 .redaction import redact_urls_in_text
13__all__ = ["exception_to_dict", "exception_to_json"]
15_MAX_VALUE_DEPTH = 8
16_NOT_SCALAR = object()
17_EXCEPTION_GROUP_TYPE = getattr(builtins, "BaseExceptionGroup", None)
20def _redact_export_text(text: str) -> str:
21 """Scrub URLs for export, including their paths.
23 Normal DataExcept messages preserve URL paths because paths are commonly
24 useful debugging context. Structured envelopes have a stricter boundary:
25 third-party errors and arbitrary caller state may put credentials in the
26 path itself, so exported text never preserves URL paths.
27 """
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)): 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true
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)): 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true
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)): 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true
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 _exception_record(
136 exc: BaseException,
137 *,
138 include_attributes: bool,
139 max_depth: int,
140 depth: int,
141 seen: set[int],
142) -> dict[str, Any]:
143 if depth > max_depth:
144 return {"truncated": True}
146 identity = id(exc)
147 if identity in seen:
148 return {
149 "type": type(exc).__name__,
150 "module": type(exc).__module__,
151 "message": _safe_text(exc),
152 "cycle": True,
153 }
155 seen.add(identity)
156 record: dict[str, Any] = {
157 "type": type(exc).__name__,
158 "module": type(exc).__module__,
159 "message": _safe_text(exc),
160 }
161 if include_attributes:
162 attributes = _attributes(exc)
163 if attributes:
164 record["attributes"] = attributes
166 members = _group_members(exc)
167 if members is not None:
168 record["exceptions"] = [
169 _exception_record(
170 member,
171 include_attributes=include_attributes,
172 max_depth=max_depth,
173 depth=depth + 1,
174 seen=seen,
175 )
176 for member in members
177 ]
179 if exc.__cause__ is not None:
180 record["cause"] = _exception_record(
181 exc.__cause__,
182 include_attributes=include_attributes,
183 max_depth=max_depth,
184 depth=depth + 1,
185 seen=seen,
186 )
187 if exc.__context__ is not None and not exc.__suppress_context__:
188 record["context"] = _exception_record(
189 exc.__context__,
190 include_attributes=include_attributes,
191 max_depth=max_depth,
192 depth=depth + 1,
193 seen=seen,
194 )
196 seen.discard(identity)
197 return record
200def exception_to_dict(
201 exc: BaseException,
202 *,
203 include_attributes: bool = True,
204 max_depth: int = 8,
205) -> dict[str, Any]:
206 """Return a strict JSON-safe structured representation of *exc*.
208 The representation contains the exception type, module and rendered
209 message, optionally public instance attributes, bounded cause/context
210 chains, and on Python 3.11+ the member tree of exception groups. Traceback
211 frames and private attributes are deliberately excluded.
212 """
213 if not isinstance(exc, BaseException):
214 raise TypeError("exc must be an exception instance")
215 if not isinstance(max_depth, int) or isinstance(max_depth, bool):
216 raise TypeError("max_depth must be an integer")
217 if max_depth < 0:
218 raise ValueError("max_depth must be non-negative")
219 return _exception_record(
220 exc,
221 include_attributes=include_attributes,
222 max_depth=max_depth,
223 depth=0,
224 seen=set(),
225 )
228def exception_to_json(
229 exc: BaseException,
230 *,
231 include_attributes: bool = True,
232 max_depth: int = 8,
233 **json_kwargs: Any,
234) -> str:
235 """Return :func:`exception_to_dict` encoded as strict JSON."""
236 json_kwargs["allow_nan"] = False
237 return json.dumps(
238 exception_to_dict(
239 exc,
240 include_attributes=include_attributes,
241 max_depth=max_depth,
242 ),
243 **json_kwargs,
244 )