Skip to main content

Advanced scheduling framework

Project description


The engine to power your Python apps

Test Test coverage Package version Supported Python versions

What is it?

Rocketry is a modern statement-based scheduling framework for Python. It is simple, clean and extensive. It is suitable for small and big projects.

This is how it looks like:

from rocketry import Rocketry
from rocketry.conds import daily

app = Rocketry()

def do_daily():

if __name__ == '__main__':

Core functionalities:

  • Powerful scheduling
  • Concurrency (async, threading, multiprocess)
  • Parametrization
  • Task pipelining
  • Modifiable session also in runtime
  • Async support


Why Rocketry?

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.

Here is an example of custom conditions:

from rocketry.conds import daily, time_of_week
from pathlib import Path

def file_exists(file):
    return Path(file).exists()

@app.task(daily.after("08:00") & file_exists("myfile.csv"))
def do_work():

Rocketry is suitable for quick automation projects and for larger scale applications. It does not make assumptions of your project structure.


Install Rocketry from PyPI:

pip install rocketry

More Examples

Here are some more examples of what it can do.


from rocketry.conds import every
from rocketry.conds import hourly, daily, weekly, 
from rocketry.conds import time_of_day
from rocketry.conds import cron

@app.task(every("10 seconds"))
def do_continuously():

def do_daily_after_seven():

@app.task(hourly & time_of_day.between("22:00", "06:00"))
def do_hourly_at_night():

@app.task((weekly.on("Mon") | weekly.on("Sat")) & time_of_day.after("10:00"))
def do_twice_a_week_after_ten():

@app.task(cron("* 2 * * *"))
def do_based_on_cron():

Pipelining tasks:

from rocketry.conds import daily, after_success
from rocketry.args import Return

def do_first():
    return 'Hello World'

def do_second(arg=Return('do_first')):
    # arg contains the value of the task do_first's return
    return 'Hello Python'

Parallelizing tasks:

from rocketry.conds import daily

@app.task(daily, execution="main")
def do_unparallel():

@app.task(daily, execution="async")
async def do_async():

@app.task(daily, execution="thread")
def do_on_separate_thread():

@app.task(daily, execution="process")
def do_on_separate_process():


Read more from the documentation.

About Library

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

rocketry-2.5.1.tar.gz (327.5 kB view hashes)

Uploaded source

Built Distribution

rocketry-2.5.1-py3-none-any.whl (216.7 kB view hashes)

Uploaded py3

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Fastly Fastly CDN Google Google Object Storage and Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page