Rocketry：一个现代化的、基于语句的Python任务调度框架

Rocketry是一个现代化的、基于语句的Python任务调度框架。它简洁、清晰且功能强大，适用于小型和大型项目。不同于其他依赖于配置文件或复杂API的任务调度器，Rocketry采用声明式编程方式，使任务调度逻辑更易于理解和维护。其核心思想是将任务及其调度条件以简洁的Python代码表达出来，让开发者专注于业务逻辑而非调度细节。

Rocketry能做什么？

Rocketry提供了一套强大的功能，用于构建各种类型的定时任务：

• • 强大的调度能力: Rocketry支持多种调度策略，包括基于时间的调度（例如，每隔10秒、每天早上7点、每周一等等）、基于cron表达式的调度，以及自定义的调度条件。
• • 并发执行: Rocketry支持异步（async）、线程（thread）和多进程（process）三种并发执行模式，可以根据任务的特性选择合适的模式以优化性能。这使得Rocketry可以高效地处理大量的并发任务，避免阻塞。
• • 任务参数化: Rocketry允许为任务传递参数，使得任务更加灵活和可复用。可以传递简单的值，也可以传递复杂的对象。
• • 任务流水线: Rocketry支持任务流水线，允许将多个任务串联起来，前一个任务的输出作为后一个任务的输入，从而实现复杂的业务流程。这极大地简化了复杂任务的编排。
• • 运行时修改: Rocketry的会话可以在运行时修改，这意味着即使任务已经开始运行，也可以动态调整调度计划，以应对不断变化的需求。

如何使用Rocketry？

Rocketry的使用非常简单，只需几行代码即可创建一个定时任务。以下是一个简单的例子：

from rocketry import Rocketry
from rocketry.conds import daily

app = Rocketry()

@app.task(daily)
def do_daily():
    print("This task runs daily.")

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

这段代码定义了一个名为do_daily的函数，并将其注册为一个每天运行的任务。@app.task(daily)装饰器指定了任务的调度条件，daily表示任务每天运行一次。 app.run()启动Rocketry的调度器，开始执行任务。

Rocketry的优势：简洁易用，高度灵活

Rocketry最大的优势在于其基于语句的调度方式。这意味着你可以使用Python的语法来定义调度规则，而不是学习复杂的配置文件或API。这使得Rocketry更加易于学习和使用，即使对于没有经验的开发者也能够快速上手。

同时，Rocketry又高度灵活，支持自定义调度条件。例如，你可以根据文件的是否存在、数据库的状态等自定义条件来触发任务。这使得Rocketry可以适应各种复杂的场景。

高级用法示例

以下是一些更高级的用法示例，展示了Rocketry的强大功能：

• • 自定义条件:

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

@app.cond()
def file_exists(file):
    return Path(file).exists()

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

• • 任务流水线:

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

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

@app.task(after_success(do_first))
def do_second(arg=Return('do_first')):
    print(f"Received: {arg}")

• • 并发执行:

from rocketry.conds import daily

@app.task(daily, execution="async")
async def do_async():
    await asyncio.sleep(1) # Simulate async operation

@app.task(daily, execution="thread")
def do_on_separate_thread():
    time.sleep(1) # Simulate thread operation

总结

Rocketry是一个功能强大、易于使用且灵活的Python任务调度框架。它通过简洁的语句式编程方式，极大地简化了任务调度的复杂性，并提供丰富的功能，例如并发执行、任务流水线和自定义调度条件。无论是简单的自动化任务还是复杂的业务流程，Rocketry都能轻松胜任，是Python开发者构建可靠、高效的定时任务系统的理想选择。

项目地址：https://github.com/Miksus/rocketry