Examples¶
Example Projects¶
- Rocketry with FastAPI (and React)¶
Rocketry template that has FastAPI (REST API) integrated and also has web UI built with React.
Example Snippets¶
Minimal example:
from rocketry import Rocketry
app = Rocketry()
@app.task('daily')
def do_things():
...
if __name__ == "__main__":
app.run()
Basic example:
from rocketry import Rocketry
app = Rocketry()
@app.task('daily')
def do_things():
...
@app.task("after task 'do_things'")
def do_after_things():
...
if __name__ == "__main__":
app.run()
Intermediate example:
from rocketry import Rocketry
from rocketry.args import Return, Session, Arg, FuncArg
from rocketry.conds import daily, time_of_week, after_success
app = Rocketry()
@app.cond()
def is_foo():
"This is a custom condition"
...
return True
@app.task(daily & is_foo)
def do_daily():
"This task runs once a day when foo is true"
...
return ...
@app.task((daily.at("10:00") | daily.at("19:00")) & time_of_week.between("Mon", "Fri"),
execution="process")
def do_complex():
"This task runs on complex interval and on separate process"
...
return ...
@app.task(after_success(do_daily))
def do_after_another(arg=Return(do_daily)):
"""This task runs after 'do_daily' and it has its the
return argument as an input"""
...
@app.task(daily)
def do_with_params(arg1=FuncArg(lambda: ...), arg2=Arg("myparam")):
"""This task runs with variety of arguments"""
...
@app.task(daily, execution="thread")
def do_on_session(session=Session()):
"This task modifies the scheduling session"
# Setting a task to run
for task in session.tasks:
if task.name == "do_after_another":
task.run(arg="...")
# Call for shut down
session.shut_down()
if __name__ == "__main__":
app.params(myparam="...")
app.run()
Advanced example:
from rocketry import Rocketry
from rocketry.args import Return, Arg
app = Rocketry()
# Custom Condition
# ----------------
@app.cond('is foo')
def is_foo():
# This is a custom condition
...
return True
# Parameters
# ----------
app.params(my_arg='Hello')
@app.param('item')
def get_item():
# This is a custom condition
...
return 'world'
# Tasks
# -----
@app.task('daily', execution="process")
def do_on_process():
"This task runs once a day and runs on separate process"
...
return ...
@app.task("after task 'do_things'")
def do_pipeline(arg1=Return('do_on_process'),
arg2=Arg('item'),
arg3=Arg('my_arg')):
"""This task runs when 'do_on_process' has succeeded.
Argument 'arg1' gets the return value of 'do_on_process'
Argument 'arg2' gets the return value of function 'get_item'
Argument 'arg3' is simply the value of a session parameter 'my_arg'"""
...
@app.task('daily & is foo', execution="thread")
def do_custom():
"""This task runs once a day and when is_foo returns True
This task runs on separate thread"""
...
@app.task('(true & true) | (false & True & ~True)')
def do_complex():
"""Notice the logical expression in the task start condition"""
if __name__ == "__main__":
app.run()