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

1"""Annotation normalization and value coercion. 

2 

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. 

6 

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""" 

14 

15from __future__ import annotations 

16 

17import datetime as _datetime 

18import enum 

19import types 

20import typing 

21from dataclasses import dataclass 

22from pathlib import Path, PurePath 

23from typing import Annotated, Any 

24 

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 

28 

29_TAG_ORDER = {"bool": 0, "int": 1, "float": 2, "path": 3, "str": 4} 

30 

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} 

43 

44 

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 

57 

58 

59def _is_union(ann: Any) -> bool: 

60 return typing.get_origin(ann) in (typing.Union, getattr(types, "UnionType", ())) 

61 

62 

63def _strip_none(members: list[Any]) -> list[Any]: 

64 return [m for m in members if m is not type(None)] 

65 

66 

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] 

72 

73 

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 

80 

81 

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) 

98 

99 

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 

111 

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 

148 

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 } 

158 

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 ) 

182 

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) 

186 

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 

197 

198 return Peeled(False, ann, completer, is_nosplit, **markers) # plain scalar 

199 

200 

201def is_flag(element: Any) -> bool: 

202 return element is bool 

203 

204 

205def sort_tags(tags: list[str]) -> list[str]: 

206 return sorted(dict.fromkeys(tags), key=lambda t: _TAG_ORDER.get(t, 99)) 

207 

208 

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) 

219 

220 

221_TYPE_PHRASE = { 

222 "bool": "true or false", 

223 "int": "an integer", 

224 "float": "a number", 

225 "path": "a path", 

226 "str": "text", 

227} 

228 

229 

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) 

233 

234 

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 

245 

246 

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 

256 

257 

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 

270 

271 

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 

294 

295 

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) 

316 

317 

318def _coerce_union(value: str, element: Any) -> Any: 

319 """Coerce a token to the best-matching member of a union (best effort). 

320 

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 

353 

354 

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. 

358 

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) 

373 

374 

375def coerce_custom(value: str, element: Any) -> Any: 

376 """Coerce to a type footman doesn't special-case, via its constructor. 

377 

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