Rocketry，一个超强的python库

大家好，我是木木。今天给大家分享一个超强的 Python 库

Rocketry 是一个现代Python任务调度库，它允许开发者以极简的方式定义和执行定时任务。这个库特别适合需要在Python环境中自动化执行重复任务的场景，如数据抓取、定时报告、系统维护操作等。Rocketry 的设计哲学强调简洁性和灵活性，使得任务调度变得既容易又强大。

特点

简易的任务定义使用Rocketry，你可以通过简单的Python装饰器来定义任务。这种方式使得阅读和编写任务代码变得非常直观。
灵活的调度选项 Rocketry 支持多种调度条件，包括基于时间的调度（如每天、每周）、基于事件的调度，甚至是自定义的复杂逻辑。
强大的错误处理它提供了详细的错误记录和重试机制，可以帮助开发者有效地管理和调试任务。

最佳实践

安装方法

Rocketry 可以通过pip轻松安装。在你的Python环境中运行以下命令即可安装：

pip install rocketry

功能示例

定义一个简单的任务下面的代码示例展示了如何定义一个每天执行一次的简单任务：

from rocketry import Rocketry
from rocketry.conds import daily
app = Rocketry()

@app.task(daily)
def my_task():
    print("执行每日任务")

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

条件复合调度以下示例展示了如何结合使用时间和条件来更精确地控制任务的执行：

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():
    ...

这些代码示例可以直接运行，展示了Rocketry在定义和调度任务时的简洁和灵活性。

深入应用

利用参数和环境状态

Rocketry 允许在任务中使用参数和环境状态，从而使得任务可以根据运行时的条件动态调整其行为。以下是一个示例，展示了如何在任务中使用环境变量来影响任务执行：

pythonCopy code
from rocketry.args import Arg

@app.task(daily)
def my_parametrized_task(my_param=Arg("my_param")):
    print(f"参数值: {my_param}")

if name == "__main__":
    app.config["my_param"] = "某个值"
    app.run()

这个功能特别适用于需要根据不同的运行环境或配置执行不同逻辑的任务。通过使用Rocketry，你可以轻松构建出灵活、可配置的自动化任务系统。

流水线任务，场景可以是支付、任务流、CI/CD等

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

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

# 在成功执行do_first后进行
@app.task(after_success(do_first))
def do_second(arg=Return('do_first')):
    ...
    return 'Hello Python'

并行任务

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():
    ...