Miksus/red-engine：Python的调度库

from redengine import RedEngine

app = RedEngine()

@app.task('daily')
def do_daily():
    ...

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

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

@app.task("daily after 07:00")
def do_daily_after_seven():
    ...

@app.task("hourly & time of day between 22:00 and 06:00")
def do_hourly_at_night():
    ...

@app.task("(weekly on Monday | weekly on Saturday) & time of day after 10:00")
def do_twice_a_week_after_morning():

from redengine.args import Return

@app.task("daily after 07:00")
def do_first():
    ...
    return 'Hello World'

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

@app.task("after tasks 'do_first', 'do_second'")
def do_after_multiple():
    # This runs when both 'do_first'
    # and 'do_second' succeed
    ...

from redengine import RedEngine
from redengine.args import Arg, Session

app = RedEngine()

# A custom condition
@app.cond('is foo')
def is_foo():
    return True or False

# A session wide parameter
@app.param('myparam')
def get_item():
    return "Hello World"


# Some example tasks
@app.task('daily & is foo', execution="process")
def do_on_separate_process(arg=Arg('myparam'))):
    "This task runs on separate process and takes a session wide argument"
    ...

@app.task("task 'do_on_separate_process' failed today", execution="thread")
def manipulate_runtime(session=Session())):
    "This task manipulate the runtime environment on separate thread"

    for task in session.tasks:
        task.disabled = True
    
    session.restart()


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