Coverage for src/footman/coerce.py: 95%
236 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-24 13:38 +0000
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-24 13:38 +0000
1"""Annotation normalization and value coercion.
3The manifest (introspection), the splitter (validation), and the executor
4(binding) all reason about a parameter's type through this one module, so a
5parameter's CLI shape is derived in exactly one place.
7A parameter is normalized by `peel` into `(multiple, element, completer)`
8and its scalar *element* is described as ordered "type tags"
9(`bool`/`int`/`float`/`path`/`str`) or as choices (`Literal`/`Enum`).
10Coercion tries the tags in **specificity order** — the most restrictive parser
11first, `str` last as the universal fallback — so `str | int` turns `"5"`
12into `5` and `"x"` into `"x"`.
13"""
15from __future__ import annotations
17import datetime as _datetime
18import enum
19import types
20import typing
21from dataclasses import dataclass
22from pathlib import Path, PurePath
23from typing import Annotated, Any
25from footman.params import _PathRequirement, ask, between, check, doc, env, suggest
26from footman.params import forward as _FORWARD
27from footman.params import nosplit as _NOSPLIT
29_TAG_ORDER = {"bool": 0, "int": 1, "float": 2, "path": 3, "str": 4}
31# The tokens a non-flag `bool` accepts (a scalar `bool` is a --flag and never
32# parses a token; these cover bool inside collections, dict values, and unions).
33_BOOL_TOKENS = {
34 "true": True,
35 "1": True,
36 "yes": True,
37 "on": True,
38 "false": False,
39 "0": False,
40 "no": False,
41 "off": False,
42}
45def _tag_of(t: Any) -> str | None:
46 if t is bool:
47 return "bool"
48 if t is int:
49 return "int"
50 if t is float:
51 return "float"
52 if isinstance(t, type) and issubclass(t, PurePath):
53 return "path"
54 if t is str:
55 return "str"
56 return None
59def _is_union(ann: Any) -> bool:
60 return typing.get_origin(ann) in (typing.Union, getattr(types, "UnionType", ()))
63def _strip_none(members: list[Any]) -> list[Any]:
64 return [m for m in members if m is not type(None)]
67def union_members(element: Any) -> list[Any]:
68 """Members of a union (None stripped), or `[element]` for a non-union."""
69 if _is_union(element):
70 return _strip_none(list(typing.get_args(element)))
71 return [element]
74def _union_of(parts: list[Any]) -> Any:
75 parts = list(dict.fromkeys(parts))
76 union = parts[0]
77 for part in parts[1:]: 77 ↛ 78line 77 didn't jump to line 78 because the loop on line 77 never started
78 union = union | part
79 return union
82@dataclass
83class Peeled:
84 multiple: bool # a list-valued parameter?
85 element: Any # scalar type / Union (or, for a mapping, the value type)
86 completer: suggest | None
87 nosplit: bool = False # opt OUT of comma-splitting (collections split by default)
88 mapping: bool = False # a dict[K, V] parameter?
89 key: Any = None # mapping key type
90 value_multiple: bool = False # mapping value is a list (dict[K, list[E]])
91 path_req: str | None = None # exists / file / dir requirement on a Path
92 bounds: tuple[float | None, float | None] | None = None # inclusive lo/hi
93 env: str | None = None # environment-variable fallback
94 checks: tuple[Any, ...] = () # post-coercion validators (check(fn))
95 doc: str | None = None # per-parameter help text (doc("..."))
96 ask: ask | None = None # prompt-if-missing marker (ask())
97 forward: bool = False # thread this value to dispatched tasks (forward)
100def peel(ann: Any) -> Peeled:
101 """Normalize a parameter annotation into (multiple, element, completer)."""
102 completer: suggest | None = None
103 is_nosplit = False
104 path_req: str | None = None
105 bounds: tuple[float | None, float | None] | None = None
106 env_var: str | None = None
107 checks: tuple[Any, ...] = ()
108 doc_text: str | None = None
109 ask_marker: ask | None = None
110 is_forward = False
112 # Strip Annotated and Optional wrappers in any order/nesting, e.g. both
113 # `Annotated[list[X], nosplit] | None` and `Annotated[list[X] | None, nosplit]`.
114 changed = True
115 while changed:
116 changed = False
117 if typing.get_origin(ann) is Annotated:
118 base, *meta = typing.get_args(ann)
119 for mark in meta:
120 if isinstance(mark, suggest):
121 completer = mark
122 elif mark is _NOSPLIT:
123 is_nosplit = True
124 elif isinstance(mark, _PathRequirement):
125 path_req = mark.kind
126 elif isinstance(mark, between):
127 bounds = (mark.lo, mark.hi)
128 elif isinstance(mark, range):
129 # A bare range: Python's half-open semantics, ints only.
130 bounds = (mark.start, mark.stop - 1)
131 elif isinstance(mark, env):
132 env_var = mark.var
133 elif isinstance(mark, check):
134 checks = (*checks, mark.fn)
135 elif isinstance(mark, doc):
136 doc_text = mark.text
137 elif isinstance(mark, ask):
138 ask_marker = mark
139 elif mark is _FORWARD:
140 is_forward = True
141 elif callable(mark) and not isinstance(mark, type): 141 ↛ 119line 141 didn't jump to line 119 because the condition on line 141 was always true
142 completer = suggest(mark) # a bare callable == suggest(fn)
143 ann, changed = base, True
144 elif _is_union(ann):
145 members = _strip_none(list(typing.get_args(ann)))
146 if len(members) == 1:
147 ann, changed = members[0], True
149 markers = {
150 "path_req": path_req,
151 "bounds": bounds,
152 "env": env_var,
153 "checks": checks,
154 "doc": doc_text,
155 "ask": ask_marker,
156 "forward": is_forward,
157 }
159 if ann is dict or typing.get_origin(ann) is dict: # dict[K, V] or bare dict
160 kv = typing.get_args(ann)
161 key_type = kv[0] if kv else str
162 value_type = kv[1] if len(kv) > 1 else str
163 value = peel(value_type) # recurse: value may be scalar / union / list
164 # A marker on the value type — dict[str, Annotated[int, between(1, 5)]]
165 # — applies to each value; an outer marker on the whole dict wins if
166 # both are present. (env stays outer-only; env() on a dict is a
167 # SpecError.)
168 return Peeled(
169 False,
170 value.element,
171 completer,
172 is_nosplit,
173 mapping=True,
174 key=key_type,
175 value_multiple=value.multiple,
176 path_req=path_req if path_req is not None else value.path_req,
177 bounds=bounds if bounds is not None else value.bounds,
178 env=env_var,
179 checks=(*checks, *value.checks),
180 doc=doc_text,
181 )
183 if ann is list or typing.get_origin(ann) is list: # list[X] / Many[X] / bare
184 element = (typing.get_args(ann) or (str,))[0]
185 return Peeled(True, element, completer, is_nosplit, **markers)
187 if _is_union(ann):
188 members = _strip_none(list(typing.get_args(ann)))
189 lists = [m for m in members if m is list or typing.get_origin(m) is list]
190 if lists: # list[X] | scalar... -> a list of the merged element types
191 parts: list[Any] = []
192 for lm in lists:
193 parts += list(typing.get_args(lm)) or [str]
194 parts += [m for m in members if typing.get_origin(m) is not list]
195 return Peeled(True, _union_of(parts), completer, is_nosplit, **markers)
196 return Peeled(False, ann, completer, is_nosplit, **markers) # scalar union
198 return Peeled(False, ann, completer, is_nosplit, **markers) # plain scalar
201def is_flag(element: Any) -> bool:
202 return element is bool
205def sort_tags(tags: list[str]) -> list[str]:
206 return sorted(dict.fromkeys(tags), key=lambda t: _TAG_ORDER.get(t, 99))
209def element_tags(element: Any) -> list[str]:
210 """Scalar coercion tags (specificity-sorted); empty for choice/unknown types."""
211 if _is_union(element):
212 tags = [
213 t for m in _strip_none(list(typing.get_args(element))) if (t := _tag_of(m))
214 ]
215 else:
216 tag = _tag_of(element)
217 tags = [tag] if tag else []
218 return sort_tags(tags)
221_TYPE_PHRASE = {
222 "bool": "true or false",
223 "int": "an integer",
224 "float": "a number",
225 "path": "a path",
226 "str": "text",
227}
230def type_phrase(tags: list[str]) -> str:
231 """A human phrase for a list of type tags: `['int']` -> "an integer"."""
232 return " or ".join(str(_TYPE_PHRASE.get(t, t)) for t in tags)
235def element_choices(
236 element: Any,
237) -> tuple[list[str] | None, type[enum.Enum] | None, tuple | None]:
238 """(choices as strings, Enum class, Literal values) for a choice element."""
239 if typing.get_origin(element) is typing.Literal:
240 values = typing.get_args(element)
241 return [str(v) for v in values], None, values
242 if isinstance(element, type) and issubclass(element, enum.Enum):
243 return [str(m.value) for m in element], element, None
244 return None, None, None
247def all_choices(element: Any) -> list[str] | None:
248 """Choice strings gathered across a union's Literal/Enum members (or a
249 scalar Literal/Enum); `None` if no member contributes choices."""
250 out: list[str] = []
251 for member in union_members(element):
252 member_choices, _, _ = element_choices(member)
253 if member_choices:
254 out.extend(member_choices)
255 return out or None
258def eagerly_checkable(element: Any) -> bool:
259 """Whether every union member is taggable (bool/int/float/path/str) or a
260 Literal/Enum — so the splitter can accept/reject a value up front. A member
261 like `UUID` or `Any` is not eagerly checkable; only binding can coerce it."""
262 for member in union_members(element):
263 if _tag_of(member) is not None:
264 continue
265 member_choices, _, _ = element_choices(member)
266 if member_choices is not None:
267 continue
268 return False
269 return True
272def coerce_scalar(value: str, tags: list[str]) -> tuple[bool, Any]:
273 """Try to coerce *value* to one of *tags* in specificity order."""
274 for tag in sort_tags(tags):
275 if tag == "bool":
276 if value.lower() in _BOOL_TOKENS:
277 return True, _BOOL_TOKENS[value.lower()]
278 elif tag == "int":
279 # `isascii` guards the gap between `str.isdigit` and `int()`:
280 # "²".isdigit() is true but int("²") raises.
281 digits = value[1:] if value[:1] in "+-" else value
282 if digits.isdigit() and digits.isascii():
283 return True, int(value)
284 elif tag == "float":
285 try:
286 return True, float(value)
287 except ValueError:
288 pass
289 elif tag == "path":
290 return True, Path(value)
291 elif tag == "str": 291 ↛ 274line 291 didn't jump to line 274 because the condition on line 291 was always true
292 return True, value
293 return False, None
296def coerce_one(value: str, element: Any) -> Any:
297 """Coerce a single token to its annotated element type (best effort)."""
298 _, enum_cls, literal = element_choices(element)
299 if enum_cls is not None:
300 for member in enum_cls: 300 ↛ 303line 300 didn't jump to line 303 because the loop on line 300 didn't complete
301 if str(member.value) == value or member.name == value:
302 return member
303 return enum_cls(value)
304 if literal is not None:
305 for lit in literal: 305 ↛ 308line 305 didn't jump to line 308 because the loop on line 305 didn't complete
306 if str(lit) == value:
307 return lit
308 return value
309 if _is_union(element):
310 return _coerce_union(value, element)
311 tags = element_tags(element)
312 if tags:
313 ok, out = coerce_scalar(value, tags)
314 return out if ok else value
315 return coerce_custom(value, element)
318def _coerce_union(value: str, element: Any) -> Any:
319 """Coerce a token to the best-matching member of a union (best effort).
321 Order: an exact Literal/Enum member (so `Literal[5] | str` yields the int
322 5, not "5"), then scalar tags in specificity order, then any custom-type
323 member's constructor (so `UUID | int` binds a real UUID); falls back to the
324 raw string when nothing matches.
325 """
326 members = union_members(element)
327 for member in members:
328 _, enum_cls, literal = element_choices(member)
329 if enum_cls is not None:
330 for m in enum_cls: 330 ↛ 327line 330 didn't jump to line 327 because the loop on line 330 didn't complete
331 if str(m.value) == value or m.name == value: 331 ↛ 330line 331 didn't jump to line 330 because the condition on line 331 was always true
332 return m
333 elif literal is not None:
334 for lit in literal:
335 if str(lit) == value:
336 return lit
337 tags = element_tags(element)
338 if tags: 338 ↛ 342line 338 didn't jump to line 342 because the condition on line 338 was always true
339 ok, out = coerce_scalar(value, tags)
340 if ok:
341 return out
342 for member in members: 342 ↛ 352line 342 didn't jump to line 352 because the loop on line 342 didn't complete
343 if ( 343 ↛ 342line 343 didn't jump to line 342 because the condition on line 343 was always true
344 isinstance(member, type)
345 and _tag_of(member) is None
346 and not issubclass(member, enum.Enum)
347 ):
348 try:
349 return coerce_custom(value, member)
350 except ValueError:
351 continue
352 return value
355def coerce_token(value: str, element: Any) -> Any:
356 """Strict `coerce_one` for a token the splitter never validated — an env
357 fallback or a `--` passthrough value.
359 Raises `ValueError` when a purely tag-typed element cannot parse the token
360 (e.g. `JOBS=abc` for an `int`), rather than passing the raw string through
361 the way `coerce_one` does for CLI tokens the splitter already validated.
362 Choice and
363 custom-type membership are left to `coerce_one` (and the caller's own
364 choices check), so union values keep working.
365 """
366 if element_tags(element) and all_choices(element) is None:
367 tags = element_tags(element)
368 ok, out = coerce_scalar(value, tags)
369 if not ok:
370 raise ValueError(f"expects {type_phrase(tags)} (got {value!r})")
371 return out
372 return coerce_one(value, element)
375def coerce_custom(value: str, element: Any) -> Any:
376 """Coerce to a type footman doesn't special-case, via its constructor.
378 Covers `UUID`, `Decimal`, and any user type whose constructor accepts a
379 string; `datetime`/`date` use `fromisoformat`. Validated here at
380 execution time (the splitter only ever sees strings), and raises
381 `ValueError` on a bad value so footman can report it cleanly.
382 """
383 # `Any`/`object` deliberately mean "take the raw string": both are classes
384 # on Python >=3.11, so without this guard `Any("x")` would raise.
385 if element is Any or element is object or not isinstance(element, type):
386 return value
387 try:
388 if issubclass(element, _datetime.datetime):
389 return element.fromisoformat(value)
390 if issubclass(element, _datetime.date): 390 ↛ 391line 390 didn't jump to line 391 because the condition on line 390 was never true
391 return element.fromisoformat(value)
392 return element(value)
393 except (ValueError, TypeError) as exc:
394 raise ValueError(f"{value!r} is not a valid {element.__name__}") from exc