使用python-telegram-bot预发布版遇协程未等待，如何同步处理？

安装python-telegram-bot预发布版本后，原有同步代码失效，该版本要求所有操作必须异步实现。我的场景是由非Telegram相关代码决定mymsg的值，无需使用CommandHandler，只需在业务逻辑中自行触发向Telegram频道发送消息。

现有代码添加async和await后，出现警告：

RuntimeWarning: coroutine 'teleg_mail' was never awaited

原因是调用异步函数teleg_mail的main是同步函数，无法直接使用await关键字。原代码如下：

async def teleg_mail(msg):
    bot = telegram.Bot('TOKEN')
    keyboard = [[InlineKeyboardButton("Publish", callback_data='1///'+msg)]]
    reply_markup = InlineKeyboardMarkup(keyboard)
    await bot.send_message('CHANNEL',msg, reply_markup=reply_markup)

def main() -> None:
    #...
    mymsg="code above will decide what's in this string"
    teleg_mail(mymsg)

if __name__ == "__main__":
    main()

解决方案 #

方法1：在同步函数中用asyncio.run()执行异步协程

直接通过asyncio.run()来运行异步函数teleg_mail，需要先导入asyncio库：

import asyncio
from telegram import Bot, InlineKeyboardButton, InlineKeyboardMarkup

async def teleg_mail(msg):
    bot = Bot('TOKEN')
    keyboard = [[InlineKeyboardButton("Publish", callback_data='1///'+msg)]]
    reply_markup = InlineKeyboardMarkup(keyboard)
    await bot.send_message('CHANNEL', msg, reply_markup=reply_markup)

def main() -> None:
    # ... 其他非异步业务逻辑
    mymsg = "code above will decide what's in this string"
    asyncio.run(teleg_mail(mymsg))

if __name__ == "__main__":
    main()

方法2：将main改为异步函数

如果后续main中可能加入更多异步操作，直接把main改成异步函数，再用asyncio.run()启动：

import asyncio
from telegram import Bot, InlineKeyboardButton, InlineKeyboardMarkup

async def teleg_mail(msg):
    bot = Bot('TOKEN')
    keyboard = [[InlineKeyboardButton("Publish", callback_data='1///'+msg)]]
    reply_markup = InlineKeyboardMarkup(keyboard)
    await bot.send_message('CHANNEL', msg, reply_markup=reply_markup)

async def main() -> None:
    # ... 其他业务逻辑（支持异步）
    mymsg = "code above will decide what's in this string"
    await teleg_mail(mymsg)

if __name__ == "__main__":
    asyncio.run(main())

优化建议 #

不要每次调用teleg_mail都创建新的Bot实例，重复创建会浪费连接资源。可以提前初始化Bot并复用：

import asyncio
from telegram import Bot, InlineKeyboardButton, InlineKeyboardMarkup

# 全局初始化Bot实例，复用连接
bot = Bot('TOKEN')

async def teleg_mail(msg):
    keyboard = [[InlineKeyboardButton("Publish", callback_data='1///'+msg)]]
    reply_markup = InlineKeyboardMarkup(keyboard)
    await bot.send_message('CHANNEL', msg, reply_markup=reply_markup)

def main() -> None:
    mymsg = "code above will decide what's in this string"
    asyncio.run(teleg_mail(mymsg))

if __name__ == "__main__":
    main()

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