Coverage for src/footman/manifest.py: 97%
211 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-24 13:38 +0000
1"""Manifest generation and caching — the "cold" path.
3The manifest is a JSON description of the command tree (groups, tasks, and the
4CLI shape of every parameter). The execution path imports the user's tasks
5module anyway, so introspecting the tree and rewriting the cache is effectively
6free. The completion hot path (`footman._complete`) only ever *reads* the
7cached JSON — it never imports this module or the user's code.
9Parameter mapping (function signature -> CLI shape):
11| Signature | CLI shape |
12| ------------------------ | ----------------------------------------- |
13| `fix: bool = False` | flag `--fix` / `--no-fix` |
14| `mode: str = "loose"` | option `--mode VALUE` |
15| `env: Literal[...]` | completable, eagerly-validated choices |
16| `count: int = 100` | typed option, validated at parse time |
17| `paths: list[Path] = ()` | repeatable option (`--paths a --paths b`) |
18| `template: Path` | required positional (exact arity) |
19| `*cmd: str` | variadic trailing passthrough |
20"""
22from __future__ import annotations
24import hashlib
25import inspect
26import json
27import os
28import warnings
29from pathlib import Path
30from typing import Any
32from footman import _describe, _paths, coerce, discover, docstrings, registry
33from footman.context import context_param_name
34from footman.params import suggest
35from footman.registry import Group
37SCHEMA_VERSION = 1
40class ManifestError(Exception):
41 """A tasks file describes a command surface footman cannot honour.
43 Raised at manifest-build time (the execution path) with a taught message;
44 the app layer reports it and exits 2.
45 """
48class CompleterError(ManifestError):
49 """A strict dynamic completer failed while refreshing its choices.
51 Raised so a broken completer surfaces as a taught error instead of
52 silently baking an empty choice list — which would disable the very
53 validation `strict=True` promises.
54 """
57class SpecError(ManifestError):
58 """A parameter's markers are inconsistent (e.g. `env()` with no default)."""
61def resolved_signature(fn: Any) -> inspect.Signature:
62 """Signature of *fn* with string annotations evaluated to real types.
64 `from __future__ import annotations` (and any PEP 563 usage) turns a
65 tasks file's annotations into strings; `eval_str` turns them back into the
66 types the grammar reasons about. Falls back to the raw signature if a name
67 cannot be resolved (e.g. a type defined in a local scope).
68 """
69 try:
70 return inspect.signature(fn, eval_str=True)
71 except (NameError, TypeError, AttributeError):
72 return inspect.signature(fn)
75def param_spec(param: inspect.Parameter) -> dict[str, Any]:
76 """Map one function parameter to its CLI shape (one manifest entry).
78 Dynamic-completer params get a transient `_completer` key that
79 `_finish` replaces with the completer's (cached) choices.
80 """
81 spec: dict[str, Any] = {"name": registry.cli_name(param.name)}
82 ann = param.annotation
83 empty = inspect.Parameter.empty
85 if param.kind is inspect.Parameter.VAR_KEYWORD:
86 raise SpecError(
87 f"**{param.name} is not supported — declare named parameters, or "
88 f"accept KEY=VALUE pairs with a dict[str, str] parameter"
89 )
91 if param.kind is inspect.Parameter.VAR_POSITIONAL:
92 spec["kind"] = "variadic"
93 if ann is not empty:
94 peeled = coerce.peel(ann) # unwrap Annotated so markers reach the spec
95 tags = coerce.element_tags(peeled.element)
96 if tags and tags != ["str"]:
97 spec["types"] = tags
98 _marker_keys(spec, peeled, param, has_default=False)
99 return spec
101 has_default = param.default is not empty
102 if has_default:
103 # Bake the default into the manifest when it survives the JSON
104 # coercion mirror (Path → str, Enum → value, …) — an additive key for
105 # help, the catalog, and the markdown exporter. An exotic default is
106 # simply omitted, never an error.
107 ok_default, encoded = _describe.jsonable(param.default)
108 if ok_default: 108 ↛ 113line 108 didn't jump to line 113 because the condition on line 108 was always true
109 spec["default"] = encoded
110 # A keyword-only parameter (after `*` or `*args`) is an option by
111 # Python's own declaration — defaultless, it is a *required* option,
112 # the same shape defaultless dicts and flags already take.
113 kw_only = param.kind is inspect.Parameter.KEYWORD_ONLY
115 if ann is empty:
116 if isinstance(param.default, bool): 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true
117 spec["kind"] = "flag"
118 elif has_default or kw_only:
119 spec["kind"] = "option"
120 if not has_default:
121 spec["required"] = True
122 else:
123 spec["kind"] = "argument"
124 return spec
126 peeled = coerce.peel(ann)
127 if peeled.mapping:
128 # A dict is always an option (--name KEY=VALUE); when it has no default
129 # it is a *required* option — footman has no positional-mapping syntax.
130 spec["kind"] = "option"
131 spec["mapping"] = True
132 if not has_default:
133 spec["required"] = True
134 _marker_keys(spec, peeled, param, has_default)
135 if peeled.nosplit:
136 spec["nosplit"] = True
137 if (ktags := coerce.element_tags(peeled.key)) and ktags != ["str"]: 137 ↛ 138line 137 didn't jump to line 138 because the condition on line 137 was never true
138 spec["key_types"] = ktags
139 vchoices = coerce.all_choices(peeled.element)
140 vtags = coerce.element_tags(peeled.element)
141 if vchoices is not None: 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true
142 spec["value_choices"] = vchoices
143 if vtags and vtags != ["str"] and coerce.eagerly_checkable(peeled.element):
144 spec["value_types"] = vtags
145 return spec
147 element = peeled.element
148 if coerce.is_flag(element) and not peeled.multiple:
149 # Only a *scalar* bool is a --flag; `list[bool]` stays a repeatable
150 # option whose tokens parse as booleans (true/false/1/0/yes/no/on/off).
151 spec["kind"] = "flag"
152 if not has_default and peeled.ask is None: # ask() prompts if absent
153 spec["required"] = True # else state it explicitly: --x or --no-x
154 _marker_keys(spec, peeled, param, has_default)
155 return spec
157 if peeled.ask is not None and not has_default:
158 # ask() makes a defaultless parameter a CLI-optional option: absence is
159 # filled by prompting (executor.bind), so the splitter must let it be
160 # missing rather than enforce it as a required positional.
161 spec["kind"] = "option"
162 elif has_default or kw_only:
163 spec["kind"] = "option"
164 if not has_default:
165 spec["required"] = True
166 else:
167 spec["kind"] = "argument"
168 _marker_keys(spec, peeled, param, has_default)
169 if peeled.multiple:
170 spec["multiple"] = True
171 if peeled.nosplit:
172 spec["nosplit"] = True
173 if peeled.completer is not None:
174 spec["dynamic"] = {"strict": peeled.completer.strict}
175 spec["choices"] = []
176 spec["_completer"] = peeled.completer
177 return spec
179 choices = coerce.all_choices(element)
180 tags = coerce.element_tags(element)
181 if choices is not None:
182 spec["choices"] = choices
183 # Emit `types` only when the element is eagerly checkable — a union with a
184 # custom member (`UUID | int`) can't be accept/rejected up front, so leave
185 # it to binding rather than eagerly rejecting valid values.
186 if tags and tags != ["str"] and coerce.eagerly_checkable(element):
187 spec["types"] = tags
188 elif choices is None and not tags and not isinstance(element, type):
189 # The annotation resolves to nothing footman can coerce (a string
190 # that never resolved, a value, an exotic generic): values will pass
191 # through as plain text. Silent degrade is a debugging tax — say so.
192 warnings.warn(
193 f"footman: parameter {param.name!r}: annotation {element!r} is "
194 f"not a usable type; values are passed through as text",
195 stacklevel=2,
196 )
197 return spec
200def _marker_keys(
201 spec: dict[str, Any],
202 peeled: coerce.Peeled,
203 param: inspect.Parameter,
204 has_default: bool,
205) -> None:
206 """Additive manifest keys for the `Annotated` markers (path/bounds/env/doc).
208 `check(fn)` deliberately never lands in the manifest — functions don't
209 serialize (the same reason `_finish` strips `_completer`); it runs at
210 binding time instead.
211 """
212 if peeled.doc is not None:
213 spec["doc"] = peeled.doc
214 if peeled.path_req is not None:
215 spec["path"] = peeled.path_req
216 if peeled.bounds is not None:
217 lo, hi = peeled.bounds
218 if lo is not None: 218 ↛ 220line 218 didn't jump to line 220 because the condition on line 218 was always true
219 spec["min"] = lo
220 if hi is not None:
221 spec["max"] = hi
222 if peeled.env is not None:
223 if spec.get("mapping"):
224 raise SpecError(
225 f"<{param.name}>: env() is not supported on dict parameters"
226 )
227 if not has_default:
228 raise SpecError(
229 f"<{param.name}>: env({peeled.env!r}) needs a default — an "
230 f"env fallback makes the parameter optional, so it needs "
231 f"somewhere to fall"
232 )
233 spec["env"] = peeled.env
236def _run_completer(completer: suggest, memo: dict[int, list[str]]) -> list[str]:
237 """Call a completer at most once per build (deduped by function identity).
239 A raising *strict* completer aborts the build with `CompleterError` — its
240 whole point is validation, so failing silent would validate nothing. A
241 best-effort completer (`strict=False`) degrades to no candidates.
242 """
243 key = id(completer.fn)
244 if key not in memo:
245 try:
246 memo[key] = [str(v) for v in completer.fn()]
247 except Exception as exc:
248 if completer.strict:
249 name = getattr(completer.fn, "__qualname__", repr(completer.fn))
250 raise CompleterError(
251 f"dynamic choices from {name}() failed: "
252 f"{type(exc).__name__}: {exc} — fix the completer, or pass "
253 f"suggest(fn, strict=False) if this data is best-effort"
254 ) from exc
255 memo[key] = []
256 return memo[key]
259def _finish(spec: dict[str, Any], memo: dict[int, list[str]]) -> dict[str, Any]:
260 completer = spec.pop("_completer", None)
261 if completer is not None:
262 spec["choices"] = _run_completer(completer, memo)
263 return spec
266def _cli_params(fn: Any):
267 """The parameters that form a task's CLI (the injected ctx is not one)."""
268 sig = resolved_signature(fn)
269 ctx_name = context_param_name(sig)
270 return [p for p in sig.parameters.values() if p.name != ctx_name]
273def _source_of(fn: Any) -> str:
274 code = getattr(fn, "__code__", None)
275 if code is None: 275 ↛ 276line 275 didn't jump to line 276 because the condition on line 275 was never true
276 return ""
277 return f"{code.co_filename}:{code.co_firstlineno}"
280def _task_node(fn: Any, memo: dict[int, list[str]]) -> dict[str, Any]:
281 sig = resolved_signature(fn)
282 infinite = registry.is_infinite(fn)
283 interactive = registry.is_interactive(fn)
284 confirm = registry.task_confirm(fn)
285 ctx_name = context_param_name(sig) # the injected ctx param is not a CLI arg
286 parsed = docstrings.parse(inspect.getdoc(fn))
287 params = [
288 _finish(param_spec(p), memo)
289 for p in sig.parameters.values()
290 if p.name != ctx_name
291 ]
292 for spec in params:
293 if spec["name"] == "help" and spec["kind"] in ("flag", "option"):
294 raise SpecError(
295 "<help>: 'help' is a reserved parameter name — it maps to "
296 "--help, which footman intercepts anywhere on the line to show "
297 "help and never run a task, so the option could never be "
298 "reached. Rename it (e.g. show_help). It is the only reserved "
299 "name: every other global must come before the first task, so a "
300 "task parameter may reuse it (fm deploy --json binds --json to "
301 "deploy)."
302 )
303 known: set[str] = set()
304 for spec in params:
305 python_name = str(spec["name"]).replace("-", "_")
306 known.add(python_name)
307 # The docstring fills in; an explicit doc() marker already won.
308 if "doc" not in spec and (text := parsed.params.get(python_name)):
309 spec["doc"] = text
310 if ctx_name:
311 known.add(ctx_name) # documenting the injected ctx param is fine
312 if unknown := sorted(set(parsed.params) - known):
313 warnings.warn(
314 f"footman: {getattr(fn, '__name__', fn)!s}: docstring documents "
315 f"unknown parameter(s): {', '.join(unknown)}",
316 stacklevel=2,
317 )
318 node: dict[str, Any] = {"help": parsed.summary, "params": params}
319 if (previous := discover.shadowed(fn)) is not None:
320 # Additive, and only for the rare overridden task: the options of
321 # the task this one shadows, so `--help` can show the call
322 # `inherited()` will make.
323 node["shadows"] = {
324 "params": [_finish(param_spec(p), memo) for p in _cli_params(previous)],
325 "where": _source_of(previous),
326 }
327 if infinite:
328 node["infinite"] = True # additive: listings and help say how it ends
329 if interactive:
330 node["interactive"] = True # additive: this task owns the terminal
331 if confirm:
332 node["confirm"] = confirm # additive: the yes/no gate before it runs
333 if parsed.long:
334 node["long"] = parsed.long
335 # Additive availability annotation (`@requires`): the name stays listed and
336 # completable either way — execution re-checks the predicate live.
337 if (reason := registry.availability(fn)) is not None:
338 node["disabled"] = reason
339 return node
342def _node(g: Group, memo: dict[int, list[str]]) -> dict[str, Any]:
343 node: dict[str, Any] = {
344 "help": g.help,
345 "tasks": {name: _task_node(fn, memo) for name, fn in g.tasks.items()},
346 "groups": {name: _node(sub, memo) for name, sub in g.groups.items()},
347 }
348 # A runnable group (one with `@group.default`) carries the default's option
349 # surface — the same `{help, params}` shape a task node has — so the splitter
350 # parses a bare `fm <group> [flags]` against it and completion/help render it.
351 if g.default_task is not None:
352 node["default"] = _task_node(g.default_task, memo)
353 return node
356def tree_hash(tree: dict[str, Any]) -> str:
357 """Stable hash of the tree's structure (names, params, help)."""
358 blob = json.dumps(tree, sort_keys=True, ensure_ascii=False)
359 return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:16]
362def build_manifest(
363 root: Group, *, completion_max_age: int | None = None
364) -> dict[str, Any]:
365 """Introspect *root* into a serialisable manifest dict.
367 Dynamic completers run here (once each, deduped) — this is the execution
368 path, so paying to refresh their cached choices is free. *completion_max_age*
369 (seconds, or `None` to disable) is baked in so the stdlib-only completion hot
370 path can decide whether to trigger a background refresh without reading config.
371 """
372 tree = _node(root, {})
373 return {
374 "schema": SCHEMA_VERSION,
375 "hash": tree_hash(tree),
376 "completion_max_age": completion_max_age,
377 "tree": tree,
378 }
381def write_manifest(manifest: dict[str, Any], path: Path) -> None:
382 """Write *manifest* to *path* atomically (never leave a half file)."""
383 path.parent.mkdir(parents=True, exist_ok=True)
384 tmp = path.with_name(f"{path.name}.{os.getpid()}.tmp")
385 tmp.write_text(json.dumps(manifest, indent=1, ensure_ascii=False), "utf-8")
386 os.replace(tmp, path)
389def load_manifest(path: Path) -> dict[str, Any] | None:
390 """Read a cached manifest, or `None` if missing/unreadable/corrupt."""
391 try:
392 data = json.loads(path.read_text("utf-8"))
393 except (OSError, ValueError):
394 return None
395 return data if isinstance(data, dict) else None
398def sync_manifest(
399 root: Group,
400 key_dir: Path,
401 *,
402 completion_max_age: int | None = None,
403 tasks_file: str | None = None,
404 path: Path | None = None,
405) -> dict[str, Any]:
406 """Build the fresh manifest and rewrite the cache only on a hash change.
408 Called on the execution path, which has already paid to import the tree.
409 The cache is keyed by *key_dir* (the cwd), since the effective task set is
410 the cascade from the repo root down. The hash guard avoids needless disk
411 writes (and mtime churn) when nothing about the command surface changed — a
412 changed *completion_max_age* also forces a rewrite so a config edit takes
413 effect.
414 """
415 fresh = build_manifest(root, completion_max_age=completion_max_age)
416 # The directory this manifest describes, baked in (additive) so the
417 # cache collector can tell a deleted project's leftovers from a living
418 # one's without guessing from hashes.
419 fresh["cwd"] = str(key_dir)
420 if tasks_file:
421 # Additive, like `cwd`: the background refresh reads it back, so a
422 # branded CLI's custom filename survives a refresh it can't attend.
423 fresh["tasks_file"] = tasks_file
424 # `path` lets a caller key the cache file separately from the baked
425 # `key_dir` — a `-f` run caches by (cwd, file) yet still bakes the cwd, so
426 # the collector prunes it with the project like any other.
427 path = path or _paths.manifest_path(key_dir)
428 cached = load_manifest(path)
429 if (
430 cached is None
431 or cached.get("hash") != fresh["hash"]
432 or cached.get("completion_max_age") != completion_max_age
433 or cached.get("cwd") != fresh["cwd"]
434 or cached.get("tasks_file") != fresh.get("tasks_file")
435 ):
436 write_manifest(fresh, path)
437 return fresh