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