There are plenty of scheduling options out there: cron, Airflow, APScheduler... However, they are all limited with the type of automation they are suitable for. Airflow was created for data engineers and cron for simple needs. There were no solutions for truly autonomous applications, until recently.

Rocketry

Rocketry is a modern scheduling framework. It was designed to be easy to use, easy to customize and it has a unique take on scheduling. It does not force you to a specific structure and it can be seamlessly integrated with other frameworks such as FastAPI.

A minimal example:

from rocketry import Rocketry
from rocketry.conds import daily

app = Rocketry()

@app.task(daily)
def do_daily():
    ... # This function runs once a day

if __name__ == "__main__":
    app.run()

Unlike the alternatives, Rocketry is a statement-based system: a task is run if its scheduling statement is true. This paradigm enables a lot of flexibility on scheduling and Rocketry natively supports cron-like scheduling, task pipelines and custom statements under the same generic mechanism. You can also arbitrarily combine the statements using basic logic: AND, OR and NOT. In contrast, Airflow has separate concepts for time scheduling, task pipelines and custom scheduling (sensors) and there are limitations on how you could combine them.

A more advanced scheduling example:

from rocketry.conds import cron, after_success

@app.cond()
def is_foo():
    ... # This is a custom condition. Should return True/False
    return True or False

@app.task(cron("0 0 * * *"))
def do_first():
    ...

@app.task((cron("10 0 * * *") & after_success(do_first)) | ~is_foo)
def do_second():
    ... 

There is a lot more than just scheduling in the framework. It also has:

• Parallelization/concurrency (async, threading, multiprocessing)
• Parameterization
• Task termination (custom or timeout)
• Keep task logs in memory, in a database or anywhere you like
• Modifiable runtime sessions (create/delete/modify tasks or restart the scheduler)

Another example to demonstrate some more features:

from rocketry.conds import daily, time_of_week
from rocketry.args import FuncArg, Return

def get_value():
    ...
    return "Hello"

@app.task(daily.after("10:00") & time_of_week.on("Mon"), execution="process")
def do_first(arg=FuncArg(get_value)):
    # "arg"s value is the return of get_value
    ...
    return "World"

@app.task(after_success(do_first), execution="async")
async def do_second(arg=Return(do_first)):
    # "arg"s value is the return of the task do_first
    ...

We used decorated functions in the examples but Rocketry supports other task types as well: from shell commands (app.task(command="echo 'Hello'")), from Python scripts (app.task(path="path/to/script.py", func_name="main")) or just passing a function (app.task(func=do_things)).

In case you got interested, here are the relevant links:

• Documentation: https://rocketry.readthedocs.io/
• Source code: https://github.com/Miksus/rocketry
• Releases: https://pypi.org/project/rocketry/

You might also like:

• (Example) Rocketry + API + web UI
• (Article) Rocketry's scheduling
• (Article) Scheduler with an API
• (Article) Initial introduction to Rocketry

Rocketry's vision is to make creating autonomous applications as easy as possible. If you liked the project, help us by spreading the word and consider leaving it a star on Github. Rocketry is also open for contribution, regardless if they are big or small. It is free open-source framework and it is developed with a passion.