Authoring in Python

Everything a build declares is a field of a class extending Build, configured with chained setters. The engine reads the fields, plans the graph and runs it; this page walks through what you can declare. Every option shown is in the Python API reference with its full signature.

The build class

from zuke import Build, run, target


class App(Build):
    clean = target().description("Remove dist/").executes(lambda ctx: ctx.sh("rm", "-rf", "dist"))
    build = target().description("Build the wheel").depends_on(clean).executes(
        lambda ctx: ctx.sh("uv", "build")
    )
    default = target().depends_on(build)


run(App)

Each field built with target() is a target, named by its field: zuke build runs clean then build. A target named default runs when no target is named. The last line, run(App), makes the file a build when the engine starts it and does nothing when the file is imported — so a test can import the class.

A body is any callable taking the context, sync or async. A lambda that returns ctx.sh(...) is awaited for you; for more than one step, write an async def:

from zuke import Build, Context, run, target


async def release(ctx: Context) -> None:
    version = (await ctx.sh("git", "describe", "--tags")).stdout.strip()
    await ctx.log(f"releasing {version}")
    await ctx.sh("uv", "publish")


class App(Build):
    test = target().executes(lambda ctx: ctx.sh("pytest", "-q"))
    publish = target().depends_on(test).executes(release)

Targets

Every setter returns the target, so they chain. The ones you will reach for most:

Setter	What it does
`.description(text)`	The line `--list` and `--help` show.
`.depends_on(*targets)`	Targets that must succeed first; each a sibling field declared above.
`.executes(body)`	The body, sync or async.
`.unlisted()`	Hide it from `--list` and `--help`.
`.requires(*parameters)`	Parameters that must have a value before it runs.
`.dry_runnable()`	Run the body under `--dry-run`, where the engine prints each command instead of running it; otherwise the body is skipped.
`.produces(*paths)`	Artifact paths it produces.

Dependencies & ordering

class App(Build):
    lint = target().executes(lambda ctx: ctx.sh("ruff", "check", "."))
    test = target().depends_on(lint).executes(lambda ctx: ctx.sh("pytest", "-q"))
    # Soft ordering: only when both are planned. A string names a later field.
    docs = target().after(test).before("publish").executes(lambda ctx: ctx.sh("mkdocs", "build"))
    # Pull notify in after this one, whenever this one runs.
    publish = target().depends_on(test).triggers("notify").executes(lambda ctx: ctx.sh("uv", "publish"))
    notify = target().always().executes(lambda ctx: ctx.log("done"))

depends_on takes the field itself. Fields initialise top to bottom, so a target may only depend on one declared above it; the engine names a forward reference as an error.
before, after and triggers also take a string, for a field declared later.
before/after only order targets that are both planned; triggers pulls its targets in.
always() runs even after the build failed; proceed_after_failure() keeps the build going past this target's failure.

Retries, timeouts & failure

class App(Build):
    flaky = target().retry(2, delay="10s").timeout("5m").executes(
        lambda ctx: ctx.sh("pytest", "-q", "tests/e2e")
    )
    rollback = target().unlisted().executes(lambda ctx: ctx.sh("./rollback.sh"))
    deploy = (
        target()
        .depends_on(flaky)
        .on_failure(rollback)       # or "fail" (the default) or "cancel-run"
        .on_cancel(rollback)        # compensate if the run is cancelled later
        .executes(lambda ctx: ctx.sh("./deploy.sh"))
    )

Durations are a string such as "250ms", "10s", "5m", "4h" or "1d", or an integer of milliseconds. retry(times, delay) runs the body up to times more times; timeout fails an attempt that runs longer. See Orchestration for how cancellation and compensation unwind a run.

Parameters

class App(Build):
    env = parameter("target environment").options("staging", "production").required()
    workers = parameter("upload workers").number().default(4)
    verbose = parameter("chatty output").boolean()
    regions = parameter("regions to deploy").array()
    api_token = parameter("deploy token").env("DEPLOY_TOKEN").secret()
    dry_run = parameter("plan only").boolean().flag("dry-run")

    deploy = (
        target()
        .requires(env, api_token)
        .executes(lambda ctx: ctx.log(f"{env.value} with {workers.value} workers"))
    )

parameter(description) starts a string parameter; .number(), .boolean() and .array() recast it, and .options(...) restricts it. The engine resolves each one from its flag, its environment variable or its default before any body runs, and .value holds the typed result — str | None, float | None, bool | None or a list — so pyright makes you handle the unset case. .secret() redacts the value from every log and record.

The flag and environment variable come from the field name the way the engine derives them for every SDK: env is --env and ENV. A snake_case name keeps its underscore in the flag — dry_run would be --dry_run — so set .flag("dry-run") when you want kebab case. See Flags and environment variables.

Groups

from zuke import group

class App(Build):
    lint = target().executes(lambda ctx: ctx.sh("ruff", "check", "."))
    types = target().executes(lambda ctx: ctx.sh("pyright"))
    test = target().executes(lambda ctx: ctx.sh("pytest", "-q"))
    # A parallel batch: the three run together, each after its own dependencies.
    checks = group(lint, types, test)

Members of a group run concurrently with one another whenever they run, each still waiting for its own dependencies — no --parallel flag needed. A group lists its members, so declare it below them. Targets outside a group stay serialized unless you pass --parallel to the whole run.

Conditions & caching

import os

class App(Build):
    build = (
        target()
        .inputs("src/**/*.py", "pyproject.toml")
        .outputs("dist/")
        .cache_key(lambda: os.environ.get("PYTHON_VERSION", ""))
        .executes(lambda ctx: ctx.sh("uv", "build"))
    )
    publish = (
        target()
        .depends_on(build)
        .condition(lambda: os.environ.get("CI") == "true")
        .executes(lambda ctx: ctx.sh("uv", "publish"))
    )

inputs and outputs fingerprint a target: when its inputs are unchanged and its outputs exist, the engine reports it cached and skips the body. cache_key adds a value of your own to the fingerprint. condition is asked just before the target's turn and skips it when it answers false. Both may be sync or async. See Caching.

Services

from zuke import Build, ServiceContext, run, service, target


async def start_db(ctx: ServiceContext) -> str:
    result = await ctx.sh("docker", "run", "-d", "-p", "5432:5432", "postgres:17")
    return result.stdout.strip()          # the handle: the container id


async def db_ready(ctx: ServiceContext, container: str) -> bool:
    probe = await ctx.sh("docker", "exec", container, "pg_isready", check=False)
    return probe.code == 0


class App(Build):
    db = (
        service()
        .description("postgres for the tests")
        .start(start_db)
        .ready(db_ready, timeout="30s")
        .stop(lambda ctx, container: ctx.sh("docker", "rm", "-f", container))
    )
    test = target().depends_on(db).executes(lambda ctx: ctx.sh("pytest", "-q"))


run(App)

A service is a target the engine starts before its dependants and stops after them. start answers a handle, which ready and stop receive; ready is polled until it answers true or its timeout passes. All three phases may run commands through the context. See Service targets.

Locks, gates & checks

async def check_health() -> None:
    ...  # raise to report a problem


class App(Build):
    env = parameter("environment").default("staging")

    deploy = (
        target()
        # One deploy per environment at a time, across every run and machine.
        .lock(lambda s: s.key(f"deploy-{env.value}").ttl("1h").wait_up_to("10m"))
        .executes(lambda ctx: ctx.sh("./deploy.sh"))
    )
    approve = (
        target()
        .depends_on(deploy)
        # Suspend the run until: zuke resume <run> --signal approved
        .waits_for(lambda s: s.external_signal("approved").timeout("72h").on_timeout("cancel-run"))
    )
    soak = (
        target()
        .depends_on(approve)
        .validate_during(lambda s: s.every("30s").check("healthy", check_health))
        .executes(lambda ctx: ctx.sh("./soak.sh"))
    )

.lock(...) holds a cross-run lock while the target runs; a key is required. See Cross-run locks.
.waits_for(...) suspends the run at the target until an external signal arrives, or until resume_when(check, every=...) answers true. See Durable run state.
.validate_during(...) runs named checks on an interval while the body runs; a check fails the target by raising.

These lambdas are evaluated after parameters resolve, so a key or a timeout may read .value, as the lock key above does.

CI files

from zuke import Build, ci, invoke, run, target


class App(Build):
    lint = target().executes(lambda ctx: ctx.sh("ruff", "check", "."))
    test = target().depends_on(lint).executes(lambda ctx: ctx.sh("pytest", "-q"))

    # zuke generate-ci writes it; zuke generate-ci --check fails on drift.
    pipeline = (
        ci()
        .path(".github/workflows/ci.yml")
        .invokes(lint, invoke(test, matrix={"os": ["ubuntu-latest", "macos-latest"]}))
    )
    # Or one job per target, with the dependencies as needs: edges.
    nightly = ci().path(".github/workflows/nightly.yml").fan_out()


run(App)

A ci() field declares a pipeline file, rendered by zuke generate-ci for GitHub Actions (the default), GitLab CI, Azure Pipelines or Bitbucket. .invokes(...) lists the targets to run as jobs, with invoke(target, ...) adjusting one; .fan_out() makes one job per target instead. Every job runs the committed launcher, so a runner needs nothing installed. See Generate your CI.

Hooks & build settings

class App(Build):
    deadline = "2h"                  # the whole run may take at most this long

    test = target().executes(lambda ctx: ctx.sh("pytest", "-q"))
    publish = target().depends_on(test).executes(lambda ctx: ctx.sh("uv", "publish"))
    unforceable = (publish,)         # --force may never re-run these

    def on_start(self) -> None:
        print("starting")            # a stray print becomes a log line

    async def on_target_end(self, target: str, status: str) -> None:
        ...

    def on_finish(self, ok: bool) -> None:
        ...

Override on_start, on_target_start, on_target_end or on_finish, sync or async; the engine calls only the ones you override. deadline bounds the whole run, and unforceable lists targets --force may never re-run.