Coverage for src/footman/_binder.py: 91%
122 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-06 02:59 +0000
1"""The document binder: a JSON value from the boundary into a typed shape.
3The inbound half of "dataclass in, dataclass out" (`_describe.json_default`
4serialises the way out). Structural rules, in priority order:
6- **Unknown keys are ignored, never refused.** A producer adds fields over
7 time; a consumer that breaks when its input gains a field is worse than
8 one that ignores it.
9- **Missing keys follow the dataclass**: a field with a default is optional,
10 a defaultless field that is absent is a taught refusal naming its path.
11- **No aliasing layer.** Keys map to field names directly.
12- **Recursion** covers nested dataclasses, `list[T]`, `dict[str, V]` and
13 `T | None`; scalar leaves reuse the one coercion pipeline, so `Path`,
14 `Literal`, enums and `datetime` behave exactly as a CLI token would.
15- **Refuse the rest, taught.** This is a binder, not a validation DSL —
16 `check(…)` owns validation, and every error names the JSON path
17 (`event.tool_input.file_path: expected text, got 3`).
18"""
20from __future__ import annotations
22import dataclasses
23import enum
24import typing
25from typing import Any
27from footman import _coerce
29_MISSING = dataclasses.MISSING
32def record_fields(target: Any) -> tuple[tuple[str, ...], frozenset[str]] | None:
33 """`(field names, the ones with no default)` for a record — the named
34 view of `_coerce.fields_of`, which is what a JSON object binds to.
36 One source so `is_document_target` and the binding branch cannot
37 disagree about what a record is. They did: a `NamedTuple` failed every
38 test in the old `is_document_target` (it is a `tuple` subclass, so
39 `get_origin` is `None`), fell through to the text path, and handed the
40 body a raw string with no warning at all.
41 """
42 fields = _coerce.fields_of(target)
43 if fields is None:
44 return None
45 return (
46 tuple(f.name for f in fields),
47 frozenset(f.name for f in fields if f.required),
48 )
51def is_document_target(ann: Any) -> bool:
52 """Whether an annotation names a shape a JSON document binds to — a
53 record (dataclass / `NamedTuple` / `TypedDict`), a `dict`, or a `list`
54 (bare or subscripted)."""
55 if record_fields(ann) is not None:
56 return True
57 origin = typing.get_origin(ann)
58 return ann in (dict, list) or origin in (dict, list)
61def _json_name(value: Any) -> str:
62 return {
63 type(None): "null",
64 bool: "true/false",
65 int: "a number",
66 float: "a number",
67 str: "text",
68 list: "an array",
69 dict: "an object",
70 }.get(type(value), type(value).__name__)
73def _strip(ann: Any) -> Any:
74 """Peel `Annotated[...]` wrappers off a field annotation (markers inside a
75 payload dataclass carry no meaning here)."""
76 while typing.get_origin(ann) is typing.Annotated: 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true
77 ann = typing.get_args(ann)[0]
78 return ann
81def _optional_member(ann: Any) -> Any:
82 """The union minus its `None` member, or `None` when *ann* admits none.
84 Any union carrying `None` admits a JSON null — `int | str | None` no
85 less than `int | None`. The old exactly-two-members reading refused
86 null for the wider shape: the union kept its other members and lost
87 its nullability. More than one member left keeps union shape, so the
88 remainder binds through the same walk.
89 """
90 if _coerce._is_union(ann):
91 members = [m for m in typing.get_args(ann) if m is not type(None)]
92 if members and len(members) < len(typing.get_args(ann)):
93 remainder = members[0]
94 for member in members[1:]:
95 remainder = remainder | member
96 return remainder
97 return None
100def _field_types(target: Any, path: str) -> dict[str, Any]:
101 try:
102 return typing.get_type_hints(target, include_extras=True)
103 except NameError as exc:
104 raise ValueError(
105 f"{path}: the annotations of {target.__name__} did not resolve "
106 f"({exc}) — a stdin payload dataclass and everything it names "
107 f"must be module-level, where `eval_str` can see them"
108 ) from exc
111def bind_document(value: Any, target: Any, path: str) -> Any:
112 """Bind one JSON *value* to *target*, recursing; *path* names where we are
113 (`event`, `event.tool_input.file_path`, `rows[2]`) so a refusal points at
114 the exact spot in the document."""
115 target = _strip(target)
117 if target is Any or target is object:
118 return value
120 member = _optional_member(target)
121 if member is not None:
122 return None if value is None else bind_document(value, member, path)
124 record = record_fields(target)
125 if record is not None:
126 names, required = record
127 if not isinstance(value, dict):
128 raise ValueError(
129 f"{path}: expected an object for {target.__name__}, "
130 f"got {_json_name(value)}"
131 )
132 hints = _field_types(target, path)
133 kwargs: dict[str, Any] = {}
134 for name in names:
135 if name in value:
136 kwargs[name] = bind_document(
137 value[name], hints.get(name, Any), f"{path}.{name}"
138 )
139 elif name in required:
140 raise ValueError(
141 f"{path}: the document has no {name!r} field and "
142 f"{target.__name__}.{name} has no default"
143 )
144 # A dataclass and a NamedTuple construct from keywords; a TypedDict
145 # called this way is simply the dict it always was at runtime.
146 return target(**kwargs)
148 origin = typing.get_origin(target)
149 if target is list or origin is list:
150 if not isinstance(value, list): 150 ↛ 151line 150 didn't jump to line 151 because the condition on line 150 was never true
151 raise ValueError(f"{path}: expected an array, got {_json_name(value)}")
152 element = (typing.get_args(target) or (Any,))[0]
153 return [
154 bind_document(item, element, f"{path}[{index}]")
155 for index, item in enumerate(value)
156 ]
158 if target is dict or origin is dict:
159 if not isinstance(value, dict): 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true
160 raise ValueError(f"{path}: expected an object, got {_json_name(value)}")
161 args = typing.get_args(target)
162 if args and _strip(args[0]) is not str: 162 ↛ 163line 162 didn't jump to line 163 because the condition on line 162 was never true
163 raise ValueError(
164 f"{path}: a JSON object's keys are text — dict keys must be "
165 f"str, not {getattr(args[0], '__name__', args[0])}"
166 )
167 element = args[1] if len(args) > 1 else Any
168 return {
169 key: bind_document(item, element, f"{path}.{key}")
170 for key, item in value.items()
171 }
173 return _leaf(value, target, path)
176def _leaf(value: Any, target: Any, path: str) -> Any:
177 """One scalar JSON value to one scalar annotation, through the same
178 pipeline a CLI token gets when the value arrives as text."""
179 if value is None or isinstance(value, (dict, list)): 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true
180 raise ValueError(f"{path}: expected {_phrase(target)}, got {_json_name(value)}")
181 if isinstance(value, str):
182 try:
183 out = _coerce.coerce_token(value, target)
184 except ValueError as exc:
185 raise ValueError(f"{path}: {exc}") from exc
186 choices = _coerce.all_choices(target)
187 if choices is not None:
188 # Choices speak tokens, so the coerced result is compared in
189 # the same face: a member's token, or the raw text for a
190 # Literal. Comparing `.value` here would refuse every member
191 # of a value-carrying enum against its own token list.
192 shown = _coerce.token_of(out) if isinstance(out, enum.Enum) else str(out)
193 if shown not in choices:
194 raise ValueError(
195 f"{path}: must be one of {'|'.join(choices)} (got {value!r})"
196 )
197 return out
198 # A JSON number or bool: no token to parse, so the fit is structural.
199 for member in _coerce.union_members(target):
200 member = _strip(member)
201 if isinstance(value, bool):
202 if member is bool:
203 return value
204 elif isinstance(value, int):
205 if member is int:
206 return value
207 if member is float:
208 return float(value)
209 elif isinstance(value, float) and member is float: 209 ↛ 211line 209 didn't jump to line 211 because the condition on line 209 was always true
210 return value
211 literal_values = (
212 typing.get_args(member)
213 if typing.get_origin(member) is typing.Literal
214 else ()
215 )
216 if any(value is lit or value == lit for lit in literal_values): 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true
217 return value
218 found = _enum_member(member, value)
219 if found is not None:
220 return found
221 raise ValueError(f"{path}: expected {_phrase(target)}, got {_json_name(value)}")
224def _enum_member(member: Any, value: Any) -> Any:
225 """The member of *member* whose value a JSON number or bool **is**, or
226 None when *member* is not an enum or nothing matches.
228 `json_default` writes an enum out as its `.value`, so a numeric-valued
229 enum arrives back as a number and has to bind — footman has to be able
230 to read the document it just wrote. Matching goes by value *and* type,
231 because `1 == True` in Python: a JSON `true` must not answer an
232 int-valued member, nor a JSON `1` a bool-valued one.
233 """
234 if not (isinstance(member, type) and issubclass(member, enum.Enum)):
235 return None
236 for candidate in member:
237 if isinstance(candidate.value, bool) != isinstance(value, bool):
238 continue
239 if candidate.value == value:
240 return candidate
241 return None
244def _phrase(target: Any) -> str:
245 tags = _coerce.element_tags(target)
246 if tags:
247 return _coerce.type_phrase(tags)
248 choices = _coerce.all_choices(target)
249 if choices: 249 ↛ 251line 249 didn't jump to line 251 because the condition on line 249 was always true
250 return "one of " + "|".join(choices)
251 return getattr(target, "__name__", str(target))