如何在基于Rocketry的项目启动时执行do_something初始化函数？

方法1：在主流程中先执行初始化任务 #

既然do_something需要依赖Pyrogram客户端启动完成，你可以调整main函数的执行顺序：先启动所有客户端，再执行初始化逻辑，最后启动Rocketry调度器（rocketry.serve()会一直运行直到调度器停止，适合放在最后执行）。

修改后的代码：

from pyrogram import Client
from rocketry import Rocketry
import asyncio

api = Client("api")
cli = Client("cli")
rocketry = Rocketry()  # 修正原代码里的变量名拼写错误：rocktry → rocketry

async def do_something():
    await api.send_message(123456, "loading ....")
    # 此处编写数据库准备等初始化代码
    await api.send_message(123456, "bot is ready!")

async def main():
    # 先启动Pyrogram客户端
    await asyncio.gather(api.start(), cli.start())
    # 执行初始化任务
    await do_something()
    # 启动Rocketry调度器（此调用会阻塞直到调度器停止）
    await rocketry.serve()

api.run(main())

方法2：用Rocketry的启动任务 #

Rocketry支持标记任务为启动时执行，通过给@rocketry.task装饰器添加startup=True参数，该任务会在调度器启动时自动执行一次，这种方式更贴合Rocketry的调度体系。

修改后的代码：

from pyrogram import Client
from rocketry import Rocketry
import asyncio

api = Client("api")
cli = Client("cli")
rocketry = Rocketry()

@rocketry.task(startup=True)
async def do_something():
    await api.send_message(123456, "loading ....")
    # 数据库准备等初始化代码
    await api.send_message(123456, "bot is ready!")

async def main():
    # 先启动Pyrogram客户端，确保初始化任务能正常调用api
    await asyncio.gather(api.start(), cli.start())
    # 启动Rocketry调度器，启动任务会自动执行
    await rocketry.serve()

api.run(main())

关键注意事项 #

• 必须保证Pyrogram客户端在do_something执行前完成启动，否则会触发客户端未初始化的错误，两种方案都先启动客户端再执行初始化，规避了这个问题。
• 原代码中存在变量名拼写错误（rocktry与rocketry.serve()不匹配），需统一为rocketry避免运行报错。

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