使用APScheduler调度Telegram Bot任务时遇Pickle序列化错误

我的代码尝试 #

调度任务代码 #

scheduler.add_job(send_delayed_message, 'date', run_date=datetime.now() + timedelta(minutes=1), kwargs={'id': order['ID'], 'bot': bot}, id=f'{order["ID"]}_first')
scheduler.add_job(send_delayed_message, 'date', run_date=datetime.now() + timedelta(minutes=3), kwargs={'id': order['ID'], 'bot': bot}, id=f'{order["ID"]}_second')

异步任务函数 #

async def send_dealayed_message(id, bot):
    bot.send_message(chat_id=id, text='Seems you didn"t come to us so far, so it"s time to do it!')

调度器初始化代码 #

scheduler = AsyncIOScheduler(timezone='Europe/Moscow')
scheduler.add_jobstore(SQLAlchemyJobStore(url='sqlite:///ProjectDB.sqlite'))

错误场景与信息 #

调度器随Telegram Bot启动，数据库已提前创建，但触发如下错误：

Task exception was never retrieved
future: <Task finished name='Task-47' coro=<Dispatcher._process_polling_updates() done, defined at C:\Users\user\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.11_qbz5n2kfra8p0\LocalCache\local-packages\Python311\site-packages\aiogram\dispatcher\dispatcher.py:407> exception=TypeError("cannot pickle '_asyncio.Task' object")>
Traceback (most recent call last):
  File "C:\Users\user\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.11_qbz5n2kfra8p0\LocalCache\local-packages\Python311\site-packages\aiogram\dispatcher\dispatcher.py", line 415, in _process_polling_updates
    for responses in itertools.chain.from_iterable(await self.process_updates(updates, fast)):
                                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\user\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.11_qbz5n2kfra8p0\LocalCache\local-packages\Python311\site-packages\aiogram\dispatcher\dispatcher.py", line 235, in process_updates
    return await asyncio.gather(*tasks)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\user\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.11_qbz5n2kfra8p0\LocalCache\local-packages\Python311\site-packages\aiogram\dispatcher\handler.py", line 117, in notify
    response = await handler_obj.handler(*args, **partial_data)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\user\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.11_qbz5n2kfra8p0\LocalCache\local-packages\Python311\site-packages\aiogram\dispatcher\dispatcher.py", line 256, in process_update
    return await self.message_handlers.notify(update.message)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\user\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.11_qbz5n2kfra8p0\LocalCache\local-packages\Python311\site-packages\aiogram\dispatcher\handler.py", line 117, in notify
    response = await handler_obj.handler(*args, **partial_data)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "c:\Users\user\projects\TelegramBot\main.py", line 199, in get_confirm
    await create_order(order, bot=bot)
  File "c:\Users\user\projects\TelegramBot\DataBase.py", line 97, in create_order
    scheduler.add_job(send_dealayed_message, 'date', run_date=datetime.now() + timedelta(minutes=1), kwargs={'id': order['ID'], 'bot': bot}, id=f'{order["ID"]}_first')
  File "C:\Users\user\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.11_qbz5n2kfra8p0\LocalCache\local-packages\Python311\site-packages\apscheduler\schedulers\base.py", line 447, in add_job
    self._real_add_job(job, jobstore, replace_existing)
  File "C:\Users\user\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.11_qbz5n2kfra8p0\LocalCache\local-packages\Python311\site-packages\apscheduler\schedulers\base.py", line 871, in _real_add_job
    store.add_job(job)
  File "C:\Users\user\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.11_qbz5n2kfra8p0\LocalCache\local-packages\Python311\site-packages\apscheduler\jobstores\sqlalchemy.py", line 97, in add_job
    'job_state': pickle.dumps(job.__getstate__(), self.pickle_protocol)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: cannot pickle '_asyncio.Task' object

问题原因 #

直接将bot对象作为参数传给调度任务，而bot内部包含_asyncio.Task这类无法被pickle序列化的对象。由于使用了SQLAlchemyJobStore，它需要将任务状态序列化后存入数据库，因此传递bot会触发序列化失败。

解决方案 #

方案1：任务内部重新初始化Bot #

抽离Bot初始化逻辑，在异步任务中重新创建Bot实例，避免传递无法序列化的对象：

async def send_delayed_message(chat_id):
    # 替换为你的Bot Token
    bot = Bot(token="YOUR_TELEGRAM_BOT_TOKEN")
    await bot.send_message(chat_id=chat_id, text='看起来你还没过来，是时候行动了！')
    await bot.session.close()  # 执行完成后关闭会话

调度任务时仅传递聊天ID：

scheduler.add_job(send_delayed_message, 'date', run_date=datetime.now() + timedelta(minutes=1), kwargs={'chat_id': order['ID']}, id=f'{order["ID"]}_first')
scheduler.add_job(send_delayed_message, 'date', run_date=datetime.now() + timedelta(minutes=3), kwargs={'chat_id': order['ID']}, id=f'{order["ID"]}_second')

方案2：使用全局Bot实例 #

如果Bot是全局初始化的，任务内部直接引用全局实例，无需作为参数传递：

# 全局初始化Bot
bot = Bot(token="YOUR_TELEGRAM_BOT_TOKEN")

async def send_delayed_message(chat_id):
    await bot.send_message(chat_id=chat_id, text='看起来你还没过来，是时候行动了！')

调度任务时同样仅传递聊天ID即可。

方案3：改用内存JobStore（不推荐） #

若不需要持久化任务，可去掉SQLAlchemyJobStore，使用默认的内存存储，避免序列化操作：

scheduler = AsyncIOScheduler(timezone='Europe/Moscow')

注意：此方案下重启服务后所有未执行的任务会丢失，仅适合临时任务场景。

另外注意：你的函数名存在拼写错误，send_dealayed_message应改为send_delayed_message，需同步修正。

内容的提问来源于stack exchange，提问作者PR. OR