Coverage for src/footman/_script.py: 95%
141 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-06 02:59 +0000
1"""PEP 723 inline script metadata — detection, and the uv command lines.
3A tasks file that carries a `# /// script` block declares its own
4dependencies, and footman runs it inside the script's uv-managed
5environment (the script handoff in `_app._uv_handoff`, and the
6self-re-exec in the completion children). This module is the one place
7that reads the block and spells the uv commands, so the run path and
8both children can never drift apart.
10Stdlib only: the completion children import it before anything heavy,
11and the block is comments — reading it never imports the file.
12"""
14from __future__ import annotations
16import contextlib
17import os
18import re
19import subprocess
20import sys
21import tomllib
22from pathlib import Path
23from typing import Any
25from footman import _paths
27# The reference regex from PEP 723, verbatim: a comment block fenced by
28# `# /// script` and `# ///`, every line a comment.
29_BLOCK = re.compile(
30 r"(?m)^# /// (?P<type>[a-zA-Z0-9-]+)$\s(?P<content>(^#(| .*)$\s)+)^# ///$"
31)
34def read_block(path: Path) -> tuple[dict[str, Any] | None, str | None]:
35 """The file's `script` metadata: `(metadata, warning)`.
37 `(None, None)` when the file has no block (or cannot be read — a
38 missing file is simply not a script). `(dict, None)` for a sound
39 block. `(None, message)` when the file *declares* a block that cannot
40 be honoured — malformed TOML, or two `script` blocks — because a
41 declared intent must never be dropped in silence: the run path prints
42 the warning and proceeds as if the block were absent.
43 """
44 try:
45 text = path.read_text(encoding="utf-8")
46 except (OSError, UnicodeDecodeError):
47 return None, None
48 blocks = [m for m in _BLOCK.finditer(text) if m.group("type") == "script"]
49 if not blocks:
50 return None, None
51 if len(blocks) > 1:
52 return None, (
53 f"{path.name} carries {len(blocks)} script blocks — PEP 723 "
54 f"allows one; running without them"
55 )
56 content = "".join(
57 line[2:] if line.startswith("# ") else line[1:]
58 for line in blocks[0].group("content").splitlines(keepends=True)
59 )
60 try:
61 return tomllib.loads(content), None
62 except tomllib.TOMLDecodeError as exc:
63 return None, (
64 f"{path.name} carries a script block footman can't read "
65 f"({exc}) — running without it"
66 )
69def _canonical(name: str) -> str:
70 """PEP 503 name normalization, after cutting the requirement's name off
71 its specifier/extras/marker tail."""
72 head = name.strip()
73 for i, ch in enumerate(head):
74 if ch in " [<>=!~;@(":
75 head = head[:i]
76 break
77 return re.sub(r"[-_.]+", "-", head).lower()
80def declares(meta: dict[str, Any], dist: str) -> bool:
81 """Whether the block's dependencies name the distribution *dist*.
83 The portability rule rides on this: a tasks file must declare the
84 runner it imports, or the script environment cannot contain it.
85 """
86 deps = meta.get("dependencies")
87 if not isinstance(deps, list):
88 return False
89 want = _canonical(dist)
90 return any(isinstance(d, str) and _canonical(d) == want for d in deps)
93def sync_argv(
94 uv: str, file: Path, *, quiet: bool = True, offline: bool = False
95) -> list[str]:
96 """`uv sync --script` — materialize (or no-op check) the script env.
98 `offline` is the completion children's mode: a keystroke never touches
99 the network, so an unmaterialized env fails fast and the child falls
100 back to running in place.
101 """
102 cmd = [uv, "sync", "--script", str(file)]
103 if quiet:
104 cmd.append("--quiet")
105 if offline:
106 cmd.append("--offline")
107 return cmd
110def find_argv(uv: str, file: Path) -> list[str]:
111 """`uv python find --script` — the script env's interpreter.
113 Only meaningful after a sync: on an unmaterialized env uv answers with
114 a *base* interpreter (exit 0), which is exactly not the script env —
115 so every caller syncs first and treats this as a lookup, not a probe.
116 """
117 return [uv, "python", "find", "--script", str(file)]
120def find_uv() -> str | None:
121 """The uv to hand off to: this runner's own environment first, then PATH.
123 Installing `footman[uv]` puts a uv binary in the same scripts directory
124 as `fm` itself, so a globally-installed runner carries its own — no
125 PATH hunt, and nothing to install separately.
126 """
127 import shutil
128 import sysconfig
130 scripts = sysconfig.get_path("scripts")
131 if scripts: 131 ↛ 135line 131 didn't jump to line 135 because the condition on line 131 was always true
132 exe = Path(scripts) / ("uv.exe" if os.name == "nt" else "uv")
133 if exe.is_file():
134 return str(exe)
135 return shutil.which("uv")
138def child_python(file: Path) -> str | None:
139 """The interpreter of *file*'s script environment — if uv can produce
140 one without the network.
142 The completion children's half of the script rule. A keystroke that
143 downloaded the world would be a broken keystroke, so the sync runs
144 `--offline`: an environment already built answers instantly, one whose
145 wheels are all in uv's cache is built there and then (no network, so
146 it is still honest), and anything else simply means "not yet" — the
147 first real run materialises it, and the TAB after that is accurate.
149 `None` means "carry on in this process", which is always safe: the
150 child either completes from what it can import, or (as it always has)
151 quietly builds nothing.
152 """
153 if os.environ.get(_paths.env_var("UV_REEXEC")) or os.environ.get(
154 _paths.env_var("NO_UV")
155 ):
156 return None
157 meta, _warning = read_block(file)
158 if meta is None or not meta.get("dependencies"):
159 return None
160 uv = find_uv()
161 if uv is None: 161 ↛ 162line 161 didn't jump to line 162 because the condition on line 161 was never true
162 return None
163 try:
164 synced = subprocess.run(
165 sync_argv(uv, file, offline=True), capture_output=True, timeout=10
166 )
167 if synced.returncode != 0:
168 return None # not materialised, and we won't reach for the network
169 found = subprocess.run(
170 find_argv(uv, file), capture_output=True, text=True, timeout=10
171 )
172 except (OSError, subprocess.SubprocessError):
173 return None
174 python = found.stdout.strip()
175 if found.returncode != 0 or not python or Path(python) == Path(sys.executable):
176 return None # already here: re-execing would only loop
177 return python
180def reexec_child(python: str, args: list[str]) -> None:
181 """Replace this completion child with the same work, run under *python*.
183 Carries the loop belt, so the re-executed child never re-enters this
184 door. Failure to exec is not an error worth making: the caller's own
185 guard swallows it and the child carries on in place.
186 """
187 os.environ[_paths.env_var("UV_REEXEC")] = "1"
188 os.environ.pop("VIRTUAL_ENV", None) # the script env is not the active one
189 os.execv(python, [python, *args])
192def maybe_reexec(files: list[Path], argv: list[str]) -> None:
193 """Continue in a script file's own environment, when one already exists —
194 the rule both completion children (`_suggest`, `_refresh`) share, kept
195 here so it cannot drift: only a single file has an environment to be
196 right about, `child_python` never touches the network (a keystroke must
197 not), and with nothing to re-exec into the caller carries on in place.
198 """
199 if len(files) != 1:
200 return # a cascade has no single environment to be right about
201 python = child_python(files[0])
202 if python is not None:
203 reexec_child(python, argv)
206# --- the lock rule's primitives, shared by the run path and the children ------
207# `_app._uv_handoff` decides where a real run belongs; the completion children
208# must reach the same verdict about the same directory, or TAB builds its
209# manifest in an environment the run would never use (a globally-installed
210# runner's own env, say) and completion silently answers from the wrong world.
213def locked_project(probe: Path) -> Path | None:
214 """The nearest ancestor holding a `uv.lock` — one existence walk, no read."""
215 return next((p for p in (probe, *probe.parents) if (p / "uv.lock").is_file()), None)
218def pins_dist(root: Path, dist: str) -> bool:
219 """Whether *root*'s lockfile pins *dist* — the question that decides
220 whether an invocation belongs to that project's environment.
222 Reading the lock is the expensive half (a real project's `uv.lock` is
223 megabytes of TOML: ~21 ms measured here), so callers ask this only once
224 the cheap answers are exhausted.
225 """
226 try:
227 with open(root / "uv.lock", "rb") as fh:
228 lock = tomllib.load(fh)
229 except (OSError, tomllib.TOMLDecodeError):
230 return False
231 return any(p.get("name") == dist for p in lock.get("package", []))
234def inside(venv: Path) -> bool:
235 """Whether this interpreter is already running out of *venv*."""
236 with contextlib.suppress(OSError):
237 return venv.is_dir() and Path(sys.prefix).resolve().is_relative_to(
238 venv.resolve()
239 )
240 return False
243def import_caused(exc: BaseException) -> bool:
244 """Whether *exc* is, or was caused by, a failed import — the shape a
245 stale environment produces, and the only one a sync can mend."""
246 seen = 0
247 cause: BaseException | None = exc
248 while cause is not None and seen < 10:
249 if isinstance(cause, ImportError):
250 return True
251 cause = cause.__cause__
252 seen += 1
253 return False
256def project_home(cwd: Path) -> Path | None:
257 """The pinned project that owns *cwd*, or None — the children's half of
258 the lock rule's question. When this answers, the script rule stays out
259 of the way, exactly as it does on the run path."""
260 root = locked_project(cwd)
261 if root is None or not pins_dist(root, _paths.dist()):
262 return None
263 return root
266def venv_python(root: Path) -> Path:
267 """Where *root*'s project environment keeps its interpreter."""
268 if os.name == "nt":
269 return root / ".venv" / "Scripts" / "python.exe"
270 return root / ".venv" / "bin" / "python"
273def heal_project(uv: str, root: Path) -> bool:
274 """Bring *root*'s environment up to its lockfile — without the network.
276 A completion child runs on a keystroke's behalf, so this is strictly
277 offline: an environment already current costs one ~15 ms check, one
278 whose missing wheels are all in uv's cache is mended there and then,
279 and anything else stays as it is — the next real run's own retry syncs
280 for real. Returns True when a sync actually ran and succeeded.
281 """
282 try:
283 fresh = subprocess.run(
284 [uv, "sync", "--check", "--quiet", "--offline", "--project", str(root)],
285 capture_output=True,
286 timeout=10,
287 )
288 if fresh.returncode == 0:
289 return False
290 synced = subprocess.run(
291 [uv, "sync", "--quiet", "--offline", "--project", str(root)],
292 capture_output=True,
293 timeout=30,
294 )
295 return synced.returncode == 0
296 except (OSError, subprocess.SubprocessError):
297 return False
300def project_reexec(root: Path, argv: list[str], *, heal: bool) -> None:
301 """Continue this completion child inside *root*'s own environment.
303 The children's spelling of `_app._uv_handoff`'s lock rule: whichever
304 runner answered the TAB, the manifest must be built by the project's
305 interpreter, from the project's packages — or completion describes a
306 world the run refuses. A foreign interpreter re-execs into the
307 project's venv; an interpreter already home stays. *heal* additionally
308 mends a stale venv offline on the way (`heal_project`) — the detached
309 refresh can afford it and asks for it unless the project opted out of
310 uv; the keystroke-facing suggest child never does. The re-exec itself
311 runs no uv at all, so the opt-out doesn't bind it.
312 """
313 if os.environ.get(_paths.env_var("UV_REEXEC")) or os.environ.get(
314 _paths.env_var("NO_UV")
315 ):
316 return
317 home = inside(root / ".venv")
318 if heal:
319 uv = find_uv()
320 if uv is not None and heal_project(uv, root) and home:
321 # Packages landed in site-packages after this interpreter first
322 # imported from it; the finders' directory caches predate them.
323 import importlib
325 importlib.invalidate_caches()
326 if home:
327 return
328 python = venv_python(root)
329 if python.is_file():
330 reexec_child(str(python), argv)