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():
...