Pipelining

There are also conditions related to if another task has runned/succeeded/failed before the task we are setting the starting condition. These are useful for creating task depenencies or task pipelines.

from rocketry.conds import after_success, after_fail, after_finish

@app.task()
def do_things():
    ...

@app.task(after_success(do_things))
def do_after_success():
    ...

@app.task(after_fail(do_things))
def do_after_fail():
    ...

@app.task(after_finish(do_things))
def do_after_fail_or_success():
    ...

You can also pipe multiple at the same time to avoid long logical statements:

from rocketry.conds import after_all_success, after_any_success, after_any_finish, after_any_fail

@app.task()
def do_a():
    ...

@app.task()
def do_b():
    ...


@app.task(after_all_success(do_a, do_b))
def do_all_succeeded():
    ...

@app.task(after_any_success(do_a, do_b))
def do_any_succeeded():
    ...

@app.task(after_any_fail(do_a, do_b))
def do_any_failed():
    ...

@app.task(after_any_finish(do_a, do_b))
def do_any_failed_or_succeeded():
    ...