No time to read? Get started then.
Rocketry is a modern scheduling framework for Python applications. It is simple, clean and extensive.
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
A lot of built-in scheduling options (including cron)
Concurrency (async, threading, multiprocessing)
Modifiable runtime session
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()
from rocketry import Rocketry app = Rocketry() @app.task('daily') def do_things(): ... if __name__ == "__main__": app.run()
There are some alternatives for a scheduler:
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.
@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(): ...
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()
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'
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')): ...
Just install the library:
pip install rocketry
- Condition Syntax
- How it works?
- Rocketry VS Alternatives
- Version history