No time to read? Get started then.
Rocketry is a modern scheduling framework for Python applications. It is simple, clean and extensive.
Key features:
Simple: Productive and easy to set up
Robust: Well tested and production ready
Extensive: A lot of built-in features
Customizable: Designed to be modified
Core functionalities:
Powerful scheduling
A lot of built-in scheduling options (including cron)
Concurrency (async, threading, multiprocessing)
Parametrization
Task pipelining
Modifiable runtime session
Async support
It looks like this:
from rocketry import Rocketry
from rocketry.conds import daily
app = Rocketry()
@app.task(daily)
def do_things():
...
if __name__ == "__main__":
app.run()
There is also a string based option
from rocketry import Rocketry
app = Rocketry()
@app.task('daily')
def do_things():
...
if __name__ == "__main__":
app.run()
Why Rocketry?¶
There are some alternatives for a scheduler:
Crontab
APScheduler
Airflow
Unlike the alternatives, Rocketry’s scheduler is statement-based. Rocketry natively supports the same scheduling strategies as the other options, including cron and task pipelining, but it can also be arbitrarily extended using custom scheduling statements.
In addition, Rocketry is very easy to use. It does not require complex setup but it can be used for bigger applications. It has a lot of options to fine-tune and a lot of features to support various needs.
Rocketry is designed to be modified and it suits well as the engine for autonomous applications. It is the automation back-end that sets your applications alive.
More Examples¶
Choosing Execution Option
@app.task(execution="main")
def do_unparallel():
...
@app.task(execution="async")
async def do_unparallel():
...
@app.task(execution="thread")
def do_on_separate_thread():
...
@app.task(execution="process")
def do_on_separate_process():
...
Custom Conditions
from pathlib import Path
from rocketry import Rocketry
from rocketry.conds import daily
app = Rocketry()
@app.cond()
def file_exists(file):
"Custom condition that checks if file exists"
return Path(file).exists()
@app.task(daily & file_exists("data.csv"))
def do_things():
"Task that runs once a day when data.csv exists"
...
if __name__ == "__main__":
app.run()
Pipelining Tasks
from rocketry.args import Return
from rocketry.conds import daily, after_success
@app.task(daily)
def do_first():
return 'Hello World'
@app.task(after_success(do_first))
def do_second(arg=Return(do_first)):
return 'Hello Python'
Parametrizing Tasks
from rocketry.args import Arg, FuncArg, EnvArg, CliArg
def get_value():
return 'Hello World'
@app.param('my_param')
def get_session_param():
"Session level parameter (named as 'my_param')"
return 'Hello Python'
@app.task()
def do_with_param(arg1=Arg('my_param'), arg2=FuncArg(get_value),
arg3=EnvArg('ENV_VARIABLE'), arg4=CliArg('--cli_arg')):
...
Interested?¶
Just install the library:
pip install rocketry
There is much more to offer. See quick start and get started with tutorials. There are also handbooks to check options and details.
Contents: