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

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

@app.task(finished(task=do_things).this_hour)
def do_if_finish():
    ...

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

@app.task(failed.this_week.on("Monday"))
def do_if_itself_fails():
    ...

Note

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

@app.task(running(do_things))
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