如何使用Rocketry编排不同虚拟环境中的Python脚本？

针对你的需求，这里提供两种可行方案，无需局限于CLI方式：

方案1：通过subprocess调用对应虚拟环境的Python解释器（推荐，隔离性好） #

将每个任务的业务逻辑单独封装成脚本，然后在Rocketry的任务函数中，调用对应虚拟环境的Python解释器执行该脚本，实现环境隔离：

1. 编写各虚拟环境对应的任务脚本：

   • 虚拟环境A的任务脚本 task_a.py：

     # 该脚本依赖需安装在virtualenv A中
     def run_daily():
         # 原do_daily任务的逻辑
         print("执行virtualenv A中的每日任务")
     if __name__ == "__main__":
         run_daily()
     

   • 虚拟环境B的任务脚本 task_b.py：

     # 该脚本依赖需安装在virtualenv B中
     def run_secondly():
         # 原do_secondly任务的逻辑
         print("执行virtualenv B中的秒级任务")
     if __name__ == "__main__":
         run_secondly()
     

2. 修改Rocketry主脚本：

   from rocketry import Rocketry
   from rocketry.conds import daily, secondly
   import subprocess
   
   app = Rocketry()
   
   @app.task(daily)
   def do_daily():
       # 替换为你的virtualenv A的Python解释器路径
       # Windows系统路径示例："C:/path/to/venv_a/Scripts/python.exe"
       venv_a_python = "/path/to/venv_a/bin/python"
       subprocess.run([venv_a_python, "task_a.py"], check=True)
   
   @app.task(secondly)
   def do_secondly():
       # 替换为你的virtualenv B的Python解释器路径
       venv_b_python = "/path/to/venv_b/bin/python"
       subprocess.run([venv_b_python, "task_b.py"], check=True)
   
   if __name__ == '__main__':
       app.run()
   

方案2：利用Rocketry的pre_task钩子切换虚拟环境（进程内切换，需注意依赖冲突） #

通过Rocketry的任务前置钩子，在任务执行前激活目标虚拟环境，让任务代码在对应环境中运行：

from rocketry import Rocketry
from rocketry.conds import daily, secondly
import os
import sys

app = Rocketry()

def activate_venv(venv_path):
    # 根据系统获取虚拟环境激活脚本路径
    if sys.platform == 'win32':
        activate_script = os.path.join(venv_path, 'Scripts', 'activate_this.py')
    else:
        activate_script = os.path.join(venv_path, 'bin', 'activate_this.py')
    # 执行激活脚本，修改当前进程的Python环境
    with open(activate_script, 'r') as f:
        exec(f.read(), {'__file__': activate_script})

@app.task(daily, pre_task=lambda: activate_venv("/path/to/venv_a"))
def do_daily():
    # 此处代码会在virtualenv A环境下运行
    import venv_a_only_module
    venv_a_only_module.execute_task()

@app.task(secondly, pre_task=lambda: activate_venv("/path/to/venv_b"))
def do_secondly():
    # 此处代码会在virtualenv B环境下运行
    import venv_b_only_module
    venv_b_only_module.execute_task()

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

注意：该方案是在同一个进程内切换虚拟环境，若两个虚拟环境存在同名但不同版本的依赖包，可能引发冲突，适合依赖无重叠的场景。

内容的提问来源于stack exchange，提问作者Felipe Messi