Bodies, context & tools

A target's body is ordinary Python. It gets one argument, ctx, which is its line to the engine: every command, every tool call, every log line and every piece of state goes through it, so the engine can apply dry-run, redaction, timeouts and cancellation the same way it does for a TypeScript build.

Bodies

from zuke import Build, Context, run, target


async def build_wheel(ctx: Context) -> None:
    if ctx.dry_run:
        await ctx.log("dry run: nothing will be uploaded")
    result = await ctx.sh("uv", "build")
    await ctx.summary({"wheel": result.stdout.strip().splitlines()[-1]})


class App(Build):
    wheel = target().executes(build_wheel)                  # an async def
    hello = target().executes(lambda ctx: ctx.log("hi"))    # a lambda returning an awaitable
    count = target().executes(lambda ctx: None)             # or plain sync code

A body may be a function, an async def or a lambda, and whatever it returns is awaited when it is awaitable — so lambda ctx: ctx.sh(...) runs the command. A body fails the target by raising. The worker's own stdout and stderr become log lines, so a stray print cannot break the run, but await ctx.log(...) is the way to say something.

Running commands

from zuke import CommandError

async def body(ctx: Context) -> None:
    out = await ctx.sh("git", "rev-parse", "HEAD")           # ExecResult(code, stdout, stderr)
    sha = out.stdout.strip()
    await ctx.sh("pytest", "-q", cwd="tests", env={"CI": "1"})
    probe = await ctx.sh("pg_isready", check=False)          # no raise; read probe.code
    try:
        await ctx.sh("false")
    except CommandError as error:
        await ctx.log(f"{error.argv} failed with {error.result.code}")

ctx.sh(*argv) asks the engine to run a command. Arguments are discrete — there is no shell, so a value with spaces or quotes is one argument and nothing can be injected. It runs in the build's directory and environment (a uv run virtualenv included); cwd, env and stdin adjust one call. A non-zero exit raises CommandError, which carries the argv and the result; pass check=False to read the code yourself. Output is streamed to the log, captured into the ExecResult, and redacted of every secret.

The context

Member	What it gives you
`await ctx.sh(*argv, ...)`	Run a command through the engine.
`await ctx.log(line)`	A log line attributed to this target.
`await ctx.secret(value)`	Redact a value from everything the run prints or records from now on.
`await ctx.summary(pairs)`	Notes in this target's row of the build summary.
`ctx.state`, `ctx.state_of(target)`	Durable state of this target, or another.
`await ctx.outcome_of(target)`, `await ctx.outcomes()`	What other targets did in this run.
`await ctx.plan()`	The planned targets and each one's dependencies.
`await ctx.signals()`	The external signals the run has received.
`ctx.run_id`, `ctx.dry_run`, `ctx.target`	The run's id, whether it is a dry run, and this target's name.

A service's phases get the smaller ServiceContext: sh, log, secret and target. Every member is in the API reference.

Durable state

VERSION = "1.4.0"


async def deploy(ctx: Context) -> None:
    state = await ctx.state.get()                 # what this target last wrote
    if state.get("deployed") == VERSION:
        return
    await ctx.sh("./deploy.sh", VERSION)
    await ctx.state.set({"deployed": VERSION})    # merged, kept across runs and resumes

    build = await ctx.outcome_of("build")         # what another target did this run
    if build is not None and build.status == "succeeded":
        ...

The engine keeps each target's state in the run record, across runs and across a suspended run's resume. set merges a patch; try_set merges it only if the engine will record it now and says whether it did. See Durable run state.

Tool facades

Every typed wrapper the engine ships — Docker, Kubernetes, Helm, Terraform, the cloud CLIs, npm, Git, GitHub, the AI reviewers and the rest — has a module under zuke.tools, generated from the engine's own tool spec, so it matches the TypeScript wrapper setter for setter:

from zuke import Build, Context, run, target
from zuke.tools.docker import DockerTasks
from zuke.tools.git import GitTasks


async def image(ctx: Context) -> None:
    sha = (await GitTasks.rev_parse(ctx, lambda s: s.rev("HEAD").short())).stdout.strip()
    await DockerTasks.build(ctx, lambda s: s.tag(f"app:{sha}").file("Dockerfile").pull())
    await DockerTasks.push(ctx, lambda s: s.image(f"app:{sha}"))

A task takes the context first, then a settings lambda. The lambda records the setters it calls, and the facade sends them to the engine in one call; the engine runs the real wrapper with your build's environment and directory, under the run's dry-run, redaction and cancellation. Pyright and your editor complete every task and setter. The tool catalog lists the wrappers — a @zuke/docker there is zuke.tools.docker here — and the Python API reference has every facade.

Reach for a facade before ctx.sh: when a tool has a zuke.tools module, its typed settings, argv purity and tool resolution come with it.

Testing a build

run(App) does nothing on import, so a build file is an ordinary module your tests can import. Each target describes itself, which makes the graph checkable without running anything:

from build import App


def test_publish_runs_after_the_tests() -> None:
    assert App.publish.manifest()["dependsOn"] == ["test"]


def test_the_targets_are_declared() -> None:
    assert list(App.targets_) == ["lint", "test", "publish"]