# pipeline
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
``` python
def PipelineError(
*args, **kwargs
):
```
*Unspecified run-time error.*
[`PipelineError`](https://vedicreader.github.io/pullup/pipeline.html#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
``` python
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
``` python
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`](https://vedicreader.github.io/pullup/pipeline.html#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`](https://vedicreader.github.io/pullup/pipeline.html#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 `, using whichever spelling exists in the
project’s `.venv/bin`. Otherwise it returns None and the pipeline falls
back to
[`resolve_script`](https://vedicreader.github.io/pullup/pipeline.html#resolve_script).
``` python
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'])
``` python
(root/'uv.lock').write_text('')
argv = uv_argv(root, ['my-tool', '-q'])
argv[0], argv[3] = Path(argv[0]).name, ''
argv
```
['uv', 'run', '--project', '', 'my_tool', '-q']
------------------------------------------------------------------------
source
### Pipeline
``` python
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`](https://vedicreader.github.io/pullup/pipeline.html#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
`//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.
``` python
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.
``` python
await p.start()
st = await settle(p)
[(r['id'], r['status'], r['code']) for r in st['steps']]
```
[('step1', 'passed', 0), ('ok', 'passed', 0)]
``` python
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.
``` python
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.
``` python
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.
``` python
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`](https://vedicreader.github.io/pullup/stack.html#blame) reads
the transcript to name which of your packages raised (see the stack
module); it returns `{}` when nothing has failed.
------------------------------------------------------------------------
source
### Release
``` python
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`](https://vedicreader.github.io/pullup/pipeline.html#release)
is the pipeline for publishing a package. Its default plan is the
project’s release flow from
[`default_steps`](https://vedicreader.github.io/pullup/project.html#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.
``` python
rl = Release(rroot)
rl.state()['flow'], [s.id for s in rl.steps], rl.path.name
```
('nbdev', ['prepare', 'bump', 'gh', 'pypi'], 'release.json')