pullup

Install

pip install pullup

Cloud integrations load only when used.

Project detection

Project detects the release flow from files in the project directory.

project = Project(root)
project.kind, [step.id for step in project.steps()]

('nbdev', ['prepare', 'bump', 'gh', 'pypi'])

Running steps

Steps run in one PTY. Automatic execution stops at the first failure.

pipeline = Pipeline(root, dir='.demo')
pipeline.save([{'id': 'hello', 'cmd': 'say hello',
                'argv': [sys.executable, '-c', 'print("ready")']}])
await pipeline.start()
pipeline.state()['done'], pipeline.tail().splitlines()[-2:]

(False, ['', '❯ say hello'])

Plan storage

Plans are written only below the project directory.

custom = Pipeline(root, dir='.myapp')
custom.save([{'id': 'show', 'cmd': 'show', 'argv': [sys.executable, '-c', 'print("saved")']}])
custom.path.relative_to(root)

Path('.myapp/pipeline.json')

Deployment pipelines

Deploy runs a project’s deployment script. Drive exposes each operation as a library call.

Deploy(root).state()['kind'], Drive(root).state()['kind']

('deploy', 'drive')

Workflows

Local workflow inspection needs no GitHub credentials.

workflow_dir = root/'.github'/'workflows'
workflow_dir.mkdir(parents=True)
(workflow_dir/'ci.yml').write_text('''name: CI
on: push
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - run: pytest
''')
workflows = Workflows(root)
[(row['name'], [job['id'] for job in row['jobs']]) for row in workflows.parsed()]

[('CI', ['test'])]

Environment

A pipeline injects required values into the child without changing the parent environment.

class Values:
    def get(self, key, secret=True): return 'demo-token' if key == 'DEMO_TOKEN' else ''
    def values(self, keys): return {key: self.get(key) for key in keys if self.get(key)}

with_values = Pipeline(root, env=Values(), dir='.env-demo')
with_values.save([{'id': 'env', 'cmd': 'read environment', 'needs': ['DEMO_TOKEN'],
                   'argv': [sys.executable, '-c', 'import os; print(os.environ["DEMO_TOKEN"])']}])
await with_values.start()
with_values.tail().splitlines()[-2:]

['', '❯ read environment']

Package errors

blame identifies the deepest selected-package frame.

trace = '''Traceback (most recent call last):
  File "/tmp/pullup/demo.py", line 7, in release
    raise RuntimeError("stopped")
RuntimeError: stopped
'''
blame(trace, family=['pullup'])

{'package': 'pullup',
 'file': '/tmp/pullup/demo.py',
 'line': 7,
 'fn': 'release',
 'error': 'RuntimeError: stopped',
 'open': '/tmp/pullup/demo.py',
 'checkout': '',
 'version': '0.0.4'}

Infrastructure

Infra.status reports provider and token availability without making a request.

infra = Infra(env={})
infra.status()

{'vpseasy': True,
 'cfeasy': True,
 'hcloud_token': True,
 'cf_token': False,
 'keys': ['HCLOUD_TOKEN', 'CLOUDFLARE_API_TOKEN']}