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 的调度框架