Miksus/red-engine:Python的调度库

22-07-04 banq

Red Engine 是 Python 应用程序的现代调度框架。它简单、干净、广泛。它是让你的 Python 程序活跃起来的引擎。
该库在表面上是最小的,但在下面是广泛且可定制的。语法非常干净:

from redengine import RedEngine

app = RedEngine()

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

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


这是一个完全工作的调度程序,它有一个每天运行一次的任务。调度语法支持 100 多个内置语句,通过逻辑(AND、OR、NOT)任意扩展它们并轻松创建自己的语句。解析引擎实际上是一个相当强大的野兽。

除了语法还有很多:
  • 持久性(任务可以记录到 CSV、SQL 或任何数据存储)
  • 并发(任务可以在不同的线程和进程上运行)
  • 流水线(执行顺序和输出到另一个输入)
  • 动态参数化(会话级和任务级)

它也有很多自定义:
  • 自定义条件
  • 自定义日志输出(即 CSV、SQL 或内存中)
  • 修改任务中的运行时环境:在常规任务中使用自定义逻辑添加任务、删除任务、修改任务、重启或关闭

它对于数据处理、爬虫、自主机器人或任何你需要安排执行代码的地方都很棒。

更多示例
调度:

@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()



Red Engine 可能具有最优雅的语法并且是最高效的。它提供了比 Crontab 或 APScheduler 更多的功能,但比 Airflow 更容易使用。它不对您的项目做出假设。为小型、中型和大型应用程序创建一个类似 FastAPI 的调度框架

猜你喜欢