Apple M1设备上aioschedule与aiogram协程调用报错求助

问题重现 #

在Apple M1设备上运行以下aiogram+aioschedule异步代码时触发错误，但Intel芯片Mac可正常运行：

import aioschedule
from aiogram import Bot, Dispatcher, executor, types

TOKEN = 'token here'
bot = Bot(TOKEN)
dp = Dispatcher(bot)

async def hello():
    print('Hello World')

async def scheduler():
    """Планувальник задач"""
    try:
        aioschedule.every().minute.do(hello)

        while True:
            await aioschedule.run_pending()
            await asyncio.sleep(5)
    except Exception as error:
        logger.error(error)

async def on_startup(_):
    """Запуск планувальника задач при старті бота"""
    asyncio.create_task(scheduler())
    logger.info('TelegramBot running...')


if __name__ == '__main__':
    try:
        executor.start_polling(dp, skip_updates=True, on_startup=on_startup)
    except Exception as err:
        logger.error(err)

错误信息：

2023-03-17 16:18:15,076 [ERROR] - __main__ - scheduler(180) - Passing coroutines is forbidden, use tasks explicitly.
/Users/emojiestore/programs/GitHub/Health-KPI/telegram_service/main.py:180: RuntimeWarning: coroutine 'Job.run' was never awaited
  logger.error(error)
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

错误原因 #

核心问题是aioschedule版本差异：

• Intel Mac上安装的旧版aioschedule允许直接将协程函数传入do()方法
• M1设备上安装的新版aioschedule（v0.6.0+）强制要求显式处理异步任务，禁止直接传递协程

解决方案 #

提供两种可行修复方式：

方式1：用asyncio.create_task包装协程调用 #

修改调度任务的注册代码，将协程调用包装为任务：

aioschedule.every().minute.do(lambda: asyncio.create_task(hello()))

方式2：使用aioschedule.async_task装饰器 #

给异步任务函数添加装饰器，标记为aioschedule可识别的异步任务：

@aioschedule.async_task
async def hello():
    print('Hello World')

# 注册任务时直接传入函数
aioschedule.every().minute.do(hello)

完整修正代码（含必要依赖导入） #

注意补充缺失的asyncio和logging导入：

import aioschedule
import asyncio
import logging
from aiogram import Bot, Dispatcher, executor, types

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

TOKEN = 'token here'
bot = Bot(TOKEN)
dp = Dispatcher(bot)

# 方式2示例：使用装饰器
@aioschedule.async_task
async def hello():
    print('Hello World')

async def scheduler():
    """Планувальник задач"""
    try:
        # 方式1示例：用lambda包装create_task
        # aioschedule.every().minute.do(lambda: asyncio.create_task(hello()))
        # 方式2示例：直接传入装饰后的函数
        aioschedule.every().minute.do(hello)

        while True:
            await aioschedule.run_pending()
            await asyncio.sleep(5)
    except Exception as error:
        logger.error(error)

async def on_startup(_):
    """Запуск планувальника задач при старті бота"""
    asyncio.create_task(scheduler())
    logger.info('TelegramBot running...')


if __name__ == '__main__':
    try:
        executor.start_polling(dp, skip_updates=True, on_startup=on_startup)
    except Exception as err:
        logger.error(err)

额外说明 #

若需统一版本，可指定安装兼容旧逻辑的aioschedule版本：

pip install aioschedule==0.5.0

但更推荐使用新版的显式任务处理方式，符合异步编程规范。

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