Task Status

Task status conditions are often useful for the end condition to prevent running a task when an important task is running or for more advanced scheduling control.

Here is a list of task status conditions:

  • rocketry.conds.started: Task has started

  • rocketry.conds.failed: Task has failed

  • rocketry.conds.succeeded: Task has succeeded

  • rocketry.conds.finished: Task has finished

All of them support periods:

  • this_minute: Status happened in the current minute (fixed)

  • this_hour: Status happened in the current hour (fixed)

  • this_day: Status happened in the current day (fixed)

  • today: Alias for this_day

  • this_week: Status happened in the current week (fixed)

  • this_month: Status happened in the current month (fixed)

All of the constrains supports additional constrains:

  • before: Status happened before given time

  • after: Status happened after given time

  • between: Status happened between given times

  • on: Status happened at given time

Here are examples:

from rocketry.conds import finished, succeeded, failed

def do_things():
    ... # Dummy task for demonstration

def do_if_finish():

@app.task(succeeded(task=do_things).today.between("10:00", "12:00"))
def do_if_fail_between():

def do_if_itself_fails():


If the task is not given, the task is interpret to be the task the condition is set to.

Task Running

There is also the condition rocketry.conds.running. This condition is true if the given task (or the task itself if not set) is running. There are also methods more_than, less_than and between to specify timespan how long the task should be running for the condition to be true:

from rocketry.conds import running

@app.task(end_cond=running.more_than("2 mins"))
def do_things():
    ... # Terminates if runs over 2 minutes

def do_if_runs():
    ... # Starts if do_things is running

@app.task(running(do_things).less_than("2 mins"))
def do_if_runs_less_than():
    ... # Starts if do_things is running less than 2 mins

@app.task(running(do_things).between("2 mins", "5 mins"))
def do_if_runs_between():
    # Starts if do_things is running
    # less than 2 mins but no more than 5 minutes

This condition can also be used to set how many parallel runs can happen (if multilaunch is set):

from rocketry.conds import running

@app.task(running <= 4, multilanch=True)
def do_parallel_limited():
    ... # Allows 4 parallel runs

@app.task(~running, multilanch=True)
def do_non_parallel():
    ... # Allows no parallel runs

@app.task(running(do_parallel_limited) >= 2)
def do_if_runs_parallel():
    ... # Runs if the other has at least two parallel runs