pipeline

a named sequence of commands for one project: the plan, the state, and the terminal it runs on

A pipeline is an ordered plan of commands for one project, together with the record of how each one went. pullup’s release, deploy and drive features are all pipelines. This module runs the commands one at a time, each in its own pseudo-terminal (PTY), copies their output into one transcript that any number of viewers can follow, and keeps the plan in a JSON file inside the project so it survives a restart.

source

PipelineError

def PipelineError(
    *args, **kwargs
):

Unspecified run-time error.

PipelineError is the only exception this module raises: for an unknown step id, a plan with nothing to run, or a request to start a step while another is running. A step that exits with a non-zero code does not raise. It is a result: the step is marked failed, its exit code is kept, and state reports both.

source

uv_argv

def uv_argv(
    root, argv
):

The command wrapped in uv run for a project with a uv.lock, or None when uv does not apply.

source

resolve_script

def resolve_script(
    argv, python:NoneType=None, env:NoneType=None
):

The command with its program replaced by a full path, trying both the hyphen and the underscore spelling.

Before running a command the pipeline has to find the program. Python tools get installed under either spelling of their name, so nbdev_prepare may be on disk as nbdev-prepare. resolve_script tries the name as given and with underscores and hyphens swapped, first in the directory of the chosen interpreter and then on PATH, and puts the full path in place of the name. A command it cannot find comes back unchanged, so the error names the program you asked for.

uv_argv handles projects managed by uv. When the project has a uv.lock and uv is installed, it returns the command wrapped in uv run --project <root>, using whichever spelling exists in the project’s .venv/bin. Otherwise it returns None and the pipeline falls back to resolve_script.

argv = resolve_script(['my-tool', '--flag'], python=bindir/'python')
Path(argv[0]).name, argv[1:], resolve_script(['not-installed-anywhere'])

('my_tool', ['--flag'], ['not-installed-anywhere'])

(root/'uv.lock').write_text('')
argv = uv_argv(root, ['my-tool', '-q'])
argv[0], argv[3] = Path(argv[0]).name, '<project>'
argv

['uv', 'run', '--project', '<project>', 'my_tool', '-q']

source

Pipeline

def Pipeline(
    root, # the project the commands run in
    python:NoneType=None, # the interpreter whose virtual environment they run in; None uses this process's
    env:NoneType=None, # an `EnvStore`, or anything with `.get(key)` and `.values(keys)`
    dir:str='.pullup', # where the plan is written, relative to `root`
):

An ordered plan of commands for one project, with each step’s result and a shared terminal transcript.

Pipeline is the class the rest of pullup builds on. You give it a project folder and, if you like, the interpreter whose virtual environment the commands should use (python), an environment store for the keys steps need (env), and the directory for the plan (dir).

save takes a list of step dictionaries and writes them to <root>/<dir>/pipeline.json. Each row needs a cmd. Rows without one are dropped, and a plan with no rows left raises. Missing ids become step1, step2 and so on. When the file is missing, unreadable or empty, the pipeline uses its class’s defaults, which for the base class is an empty plan.

state returns everything a caller needs to draw the pipeline: each step with its status (pending, running, passed, failed or skipped), its exit code and the keys it is missing, plus whether the whole plan is done.

p = Pipeline(proj, dir='.demo')
st = p.save([{'label': 'say hello', 'cmd': 'echo hello from a pipeline'},
             {'id': 'ok', 'label': 'exit clean', 'cmd': 'true'},
             {'label': 'no command of its own'}])
[(r['id'], r['label'], r['status']) for r in st['steps']]

[('step1', 'say hello', 'pending'), ('ok', 'exit clean', 'pending')]

start runs one step: the one you name, or else the first that has not passed or been skipped. With auto=True, the default, each step that passes starts the next, and the first failure ends the run. start returns once the step has been launched, so the example waits for the run to settle before reading the state. tail then returns the end of the transcript as plain text, with colour codes removed.

await p.start()
st = await settle(p)
[(r['id'], r['status'], r['code']) for r in st['steps']]

[('step1', 'passed', 0), ('ok', 'passed', 0)]

print(p.tail(9))


❯ echo hello from a pipeline
hello from a pipeline

✓ say hello exited 0

❯ true

✓ exit clean exited 0

A failing step leaves every step after it pending, and its exit code is the child process’s own. To carry on, fix the cause and call start again, which picks up at the failed step, or skip it.

q = Pipeline(proj, dir='.fail')
q.save([{'id': 'tests', 'label': 'tests', 'cmd': 'sh -c "exit 3"'},
        {'id': 'publish', 'label': 'publish', 'cmd': 'echo publishing'}])
await q.start()
[(r['id'], r['status'], r['code']) for r in (await settle(q))['steps']]

[('tests', 'failed', 3), ('publish', 'pending', None)]

A user interface usually shows the transcript live. subscribe returns an asyncio.Queue that first holds everything already in the transcript and then receives each new chunk of output as it arrives. unsubscribe stops the updates. write sends keystrokes to the running step, for example when twine asks for a password, and resize changes the size of its terminal.

sub = p.subscribe()
b'hello from a pipeline' in sub.get_nowait()

True

needs lists every environment key the plan names, once each, in the order the steps name them. missing lists the keys one step needs that the store cannot supply, so a user interface can ask for them before the step runs and fails.

r = Pipeline(proj, dir='.needs', env=Store(GITHUB_TOKEN='ghp_pretend'))
r.save([{'id': 'gh', 'label': 'GitHub release', 'cmd': 'true', 'needs': ['GITHUB_TOKEN']},
        {'id': 'pypi', 'label': 'PyPI', 'cmd': 'true', 'needs': ['TWINE_PASSWORD', 'GITHUB_TOKEN']}])
r.needs(), [(s.id, r.missing(s)) for s in r.steps]

(['GITHUB_TOKEN', 'TWINE_PASSWORD'],
 [('gh', []), ('pypi', ['TWINE_PASSWORD'])])

The remaining controls change the results, never the saved plan. skip marks a step done without running it. stop kills the running step, marks it failed and turns off automatic continuation. reset forgets every result and clears the transcript, so the plan can run again from the top. reset_plan puts the default plan back in memory and leaves the saved file alone. After a failure, blame reads the transcript to name which of your packages raised (see the stack module); it returns {} when nothing has failed.

source

Release

def Release(
    root, # the project the commands run in
    python:NoneType=None, # the interpreter whose virtual environment they run in; None uses this process's
    env:NoneType=None, # an `EnvStore`, or anything with `.get(key)` and `.values(keys)`
    dir:str='.pullup', # where the plan is written, relative to `root`
):

The release pipeline, whose default plan is the project’s release flow.

Release is the pipeline for publishing a package. Its default plan is the project’s release flow from default_steps, it saves to release.json, and its state also reports the flow’s name and whether the project is nbdev, fastship, or builds a desktop app.

rl = Release(rroot)
rl.state()['flow'], [s.id for s in rl.steps], rl.path.name

('nbdev', ['prepare', 'bump', 'gh', 'pypi'], 'release.json')

Other Formats

PipelineError

uv_argv

resolve_script

Pipeline

Release