Command line¶

adag run runs a target from a pipeline file. Every module-level StepResult in the file is a target.

"""A pipeline file for `adag run`: module-level step results become targets."""

from dataclasses import dataclass

from pydantic import BaseModel

from adag import InputParam, Step


class Number(BaseModel):
    value: int


@dataclass(frozen=True)
class Square(Step[Number]):
    def run(self, x: int) -> Number:
        return Number(value=x * x)


@dataclass(frozen=True)
class AddOne(Step[Number]):
    def run(self, n: Number) -> Number:
        return Number(value=n.value + 1)


x = InputParam[int]("x", value_type=int)
squared = Square()(x=x)
total = AddOne()(n=squared)

List the targets:
```
adag run cli_pipeline.py -h
```

List a target's params:

adag run cli_pipeline.py --target total -h

Run it:

adag run cli_pipeline.py --target total --param x=4

total: Number(value=17)

(output trimmed: log lines removed)

Add --cache-dir DIR to cache, and --tree to print the run tree.

Give a required InputParam its type with value_type=, for example InputParam[int]("x", value_type=int). The [int] alone is lost at runtime. Without value_type or a default, the param shows as <any> and the value is parsed as an int, float or bool when it looks like one, else kept as a string.

Worker command¶

adag resolve-branch runs one fan-out branch and writes its result to the cache. A dispatcher builds this command (BranchJob.command()); you rarely type it.

adag resolve-branch pkg.pipeline --fan-out per_key --key a --cache-dir /shared/cache

pkg.pipeline: a module name or a file path.
--cache-dir DIR or --store-url URL: one is required.
--params-json JSON: the run params as one JSON object.
--expect-hash HASH: exit 2 when the worker derives a different branch hash.

See Dispatch.