API reference¶

Gate a task on a live predicate — available only while it is truthy.

The generic gate the three specialisations build on. A predicate that raises reads as unavailable, the exception named:

@task
@requires(lambda: Path("config.toml").exists(), reason="needs config.toml")
def publish(): ...

Gate a task on Python modules being importable (find_spec, no import).

Keep the real import in the body; this only checks availability, so a missing optional dependency lists as a clean reason, never an import crash.

Gate a task on command-line tools being on PATH (shutil.which).

Gate a task on environment variables being set (in os.environ).

Run a command or a Python callable in the current task's context.

Subprocess output is decoded as UTF-8 by default; pass encoding= for a tool that speaks another code page, or encoding=None for the locale default. Ignored for callables (in-process, no bytes boundary).

_show is an internal channel from the tools.* bridge: a structured view of the call, so the shown command line can be normalised and role-coloured while execution runs whatever the tool needs. An explicit title still wins; a direct run([...]) is unaffected.

Run task calls / thunks concurrently; wait; fail if any fail.

Each call runs in a child of the current context with its own output buffer, flushed atomically on completion so concurrent output never interleaves. Pass task functions directly (parallel(lint, typecheck)) or thunks for arguments (parallel(lambda: build("web"), lambda: build("api"))).

Arguments after -- on the command line, for the running task.

The task this one shadows in the cascade — footman's super().

A nearer tasks.py overriding a task by name usually wants to extend it, not replace it. Call this inside the overriding task's body to get the task it shadows, then call that like the plain function it is:

# svc/api/tasks.py — the root also defines `check`
@task
def check(fix: bool = False, contracts: bool = True):
    inherited()(fix=fix)          # arguments are forwarded explicitly
    if contracts:
        run("./verify-contracts.sh")

Forwarding is deliberately manual: the two signatures are independent (a leaf usually adds a parameter), so automatic forwarding could only drop arguments silently or fail at run time — where spelling the call out shows you the mismatch as you type it. Being an ordinary call, it also runs to completion before the next statement — and composes with parallel(inherited(), extra) when you want otherwise.

fm --where <task> lists the whole shadow chain; fm --help <task> shows the inherited task's options, so you can read the forwarding call straight off it.

Report this task's own progress: done of total units.

Some work knows exactly how far along it is — 23 of 150 migrations, bytes of a download — and that is better evidence than any duration history. A reporting task's counts drive the live bar directly (counted beats estimated), so the bar is honest on the very first run, where the estimator would still be guessing.

@task
def migrate():
    for i, record in enumerate(records, 1):
        apply(record)
        progress(i, len(records))

A total of 0 (or less) clears the report, returning the run to its estimate. Outside a run, or with no live status line, this is a no-op — plain calls and captured runs cost nothing.

Iterate iterable, reporting progress as it goes.

The ergonomic form of progress(): the total comes from len() when the iterable has one, or from total when you know it for a generator. Without either, iteration still works — the run simply keeps whatever progress it had.

@task
def migrate():
    for record in track(load_records()):
        apply(record)

Ask the person running the task for a line of input.

A bare input() doesn't work in a task: its prompt goes to stdout, which footman buffers per task so parallel output never interleaves (and --json stays one envelope), so the prompt is swallowed and the task looks hung. prompt() writes to the real terminal on stderr instead — never captured — and serialises concurrent prompts.

Usable only inside an @task(interactive=True) task; called in an ordinary task body it raises a taught error naming the two fixes. Off a terminal, under --no-input, or when it would otherwise block, it returns default if given, else raises — an unattended run fails loudly. For a value a script must supply, take it as a task parameter (a CLI flag) instead.

Ask a yes/no question. --yes auto-answers yes; Enter alone takes default; off a terminal or under --no-input the answer is default. Guarded like prompt() — interactive tasks only.

Let the person pick from a runtime-computed list — the one interactive case a flag can't cover, because the options aren't known until the task runs (which changed packages to release, which stale branches to delete).

options are strings, or (label, value) pairs to show one thing and return another. multiple=True returns the chosen subset as a list; otherwise one value is returned. Guarded like prompt() (interactive tasks only), and off a terminal or under --no-input it returns default, or raises if none was given.

Download url (cached), returning the path to the local file.

A second call for the same URL revalidates with the server (ETag / Last-Modified) rather than re-downloading; a 304 Not Modified costs one round trip and keeps "cached" honest. Pass refresh to skip revalidation and fetch unconditionally.

@task
def deps():
    "Fetch the toolchain."
    archive = fetch(TOOLCHAIN_URL, sha256="9f86d0…")
    tools.tar("-xzf", archive, "-C", "vendor")

into copies the cached file to a path of your choosing (and returns that path). sha256 verifies what arrived and refuses a mismatch — the way to make a build reproducible. backend overrides the configured one for this call.

Under --dry-run nothing is downloaded: the step is recorded and the would-be cache path returned, so a plan can be inspected safely.

Fail the current task with a reason (and exit code, default 1).

The blessed way to stop a task deliberately: fail("no open PR to act on"), or fail("reserved branch", code=3) to pick the exit code too. A function, not a raise, on purpose — a task lives in your repo under your linter, and raise SomeError("a literal") trips flake8-errmsg (EM101) and tryceratops (TRY003) at the call site, every failure. A call trips neither, the same reason sys.exit() and pytest.fail() are functions. return N still spells a bare code; sys.exit(...) still works — fail() is just the footman-native one, with a reason and a code together.

Graft another module's tasks into the current tree (or into a group).

include("shared_tasks")                          # everything, at root
include("shared_tasks", only=["lint", "fmt"])    # cherry-pick by CLI name
include("mkdocs_helpers.tasks", into=docs)       # namespace: fm docs …
include(plugin("mkdocs"), only=["build"])        # from an entry point

source is a dotted module name, an imported module, or a Group. The provider imports under a registry capture, so its decorators can't leak into your tree. Collisions are loud (RegistrationError) unless override=True; unknown only=/exclude= names are errors too (typo protection). Included tasks run from your file's directory — a shared lint task lints this project. Returns the group it grafted into.

The Group a package advertises under the footman.tasks entry point.

# the plugin package's pyproject.toml
[project.entry-points."footman.tasks"]
mkdocs = "footman_mkdocs:tasks"

Raises RegistrationError naming the installed entry points when name isn't one of them — a configured-but-missing plugin should read as the typo or missing install it is.

Redirect module-level @task/group registration into a fresh tree.

The seam include() uses to import a provider module without letting its decorators land in the current registry: root.tasks/root.groups are swapped for fresh dicts for the duration and the captured tree is yielded. Reentrant — a provider may itself include() another provider.

Parse text (a raw or already-cleaned docstring) into its pieces.

One markdown document for the node at path (empty = whole tree).

A linked set of files for the node at path: index.md per group (name, help, a table of children with relative links), one file per task. Keys are POSIX-relative paths.

Install ctx as the current run context for the duration of the block.

The public seam for calling tasks from other Python code — tests included: run() and tools.* inside the block read this context instead of a fresh default. footman.testing.recording builds on it.

with use_context(Context(env={"CI": "1"})) as ctx:
    deploy()
assert ctx.steps[0].code == 0

Capture the commands a block would run() — silently, not executing.

Yields the live step list; each run()/tools.* call inside the block appends a Result instead of executing. In-process callables passed to run() are skipped too — that is the point, but worth knowing. Keyword overrides go to the underlying Context (e.g. env={...}).