Skip to main content

steps.task

The steps.task function describes a call without running it. Pass the descriptors to steps.parallel, which runs them with a concurrency limit and applies each task's retry policy and timeout.

Usage​

steps.task(name, function, args = [], kwargs = {}, retry = None, timeout = "")

Arguments​

name

Required. A non-empty label for the task. It prefixes the task's output, identifies it in errors, and must be unique within one steps.parallel group.

function
Required. The function to call, such as deploy. Pass the function itself, not a call.
args
(Optional) A list or tuple of positional arguments for the call.
kwargs
(Optional) A dictionary of keyword arguments for the call. Keys must be strings.
retry

(Optional) A dictionary describing the retry policy. See Retry policy. Without a policy a task runs once.

timeout

(Optional) A positive duration string such as "30s" or "5m". The timeout bounds the whole task, including every attempt and the delays between attempts. A task that exceeds it fails with task "<name>" timed out after <duration>.

Returns​

A task descriptor. It does nothing until it is passed to steps.parallel, and the group returns the function's return value in the matching position.

Retry policy​

The retry dictionary accepts these keys:

max_attempts
The most times the function runs. Must be positive.
backoff_strategy
One of constant, linear, or exponential.
initial_delay, max_delay, max_elapsed_time
Duration strings such as "2s". The initial delay can be zero, the others must be positive.
multiplier, random_jitter
Numbers that tune the backoff growth and add randomness to the delay.

A retry runs the entire function again, including operations that already succeeded inside it, so retry only work that is safe to repeat. A handled nonzero exit code from exec.run(..., check=False) does not retry: call fail() when the result should fail the attempt.

Set max_attempts or a timeout on every task that has a retry policy. A policy without max_attempts keeps retrying until the task succeeds, its timeout expires, or the script is canceled.

The conditions field of the Atmos retry configuration matches subprocess output and does not apply to function tasks; using it is an error, as is any key not listed above.

Errors​

  • An empty name, a non-callable function, non-string kwargs keys, an invalid timeout, or an invalid retry fails with an argument error when steps.task is called.
  • A task that still fails after its last attempt is reported with its name, the attempt count, and a traceback.

Examples​

Pass arguments and a timeout​

def deploy(name, region):
    return exec.run(["./deploy.sh", name, region], output = "capture").stdout

output = steps.parallel(
    tasks = [
        steps.task(name = "api", function = deploy, args = ["api"], kwargs = {"region": "us-east-1"}, timeout = "5m"),
        steps.task(name = "worker", function = deploy, args = ["worker", "us-west-2"], timeout = "5m"),
    ],
)

Retry a flaky operation​

def fetch_status():
    return exec.run(["curl", "--fail", "--silent", "https://status.example.com/health"], output = "capture").stdout

output = steps.parallel(
    tasks = [
        steps.task(
            name = "health",
            function = fetch_status,
            retry = {"max_attempts": 5, "initial_delay": "2s", "backoff_strategy": "exponential"},
            timeout = "2m",
        ),
    ],
)

If the command fails twice and then succeeds, the group returns the output of the third attempt.

One task per component​

def plan(component, stack):
    return atmos.terraform("plan", component, stack, output = "capture").exit_code

output = steps.parallel(
    tasks = [
        steps.task(name = c, function = plan, args = [c, "dev"], retry = {"max_attempts": 2})
        for c in ["vpc", "eks", "rds"]
    ],
    max_concurrency = 2,
    fail_fast = True,
)

Related​