如何正确运行python-telegram-bot与FastAPI？移除nest_asyncio方案

问题核心是原代码混用线程与嵌套事件循环，导致必须依赖nest_asyncio。要规范实现，需将Telegram Bot和FastAPI纳入同一异步事件循环管理，避免线程与嵌套循环。

重构后的代码 #

import asyncio
import uvicorn

from fastapi import FastAPI, APIRouter
from starlette.responses import HTMLResponse

from telegram import Update
from telegram.ext import Application, CommandHandler, CallbackContext


class WebAppWrapper:
    def __init__(self, port: int = 5000):
        self.router = APIRouter()
        self.port = port
        self.app = FastAPI()
        self.app.include_router(self.router)
        
        html_content = """
        <html>
            <head>
                <title>Some HTML in here</title>
            </head>
            <body>
                <h1>Look ma! HTML!</h1>
            </body>
        </html>
        """
        
        @self.app.get('/', include_in_schema=False)
        async def root():
            return HTMLResponse(content=html_content, status_code=200)

    async def start(self):
        config = uvicorn.Config(self.app, host="0.0.0.0", port=self.port)
        server = uvicorn.Server(config)
        await server.serve()


class TelegramBot:
    def __init__(self):
        self.bot = None
        self.web = None

    async def run(self):
        # 初始化Telegram Bot
        self.bot = Application.builder()\
            .token("SOME_TELEGRAM_BOT_TOKEN")\
            .build()

        # 添加命令处理器
        async def test_callback(update: Update, context: CallbackContext):
            await update.message.reply_text("pong")
        
        self.bot.add_handler(CommandHandler("ping", test_callback, block=False))

        # 初始化Web服务
        self.web = WebAppWrapper(port=5000)
        
        # 同时启动Web服务和Bot
        web_task = asyncio.create_task(self.web.start())
        
        # 异步管理Telegram Bot生命周期
        async with self.bot:
            await self.bot.start()
            await self.bot.updater.start_polling(drop_pending_updates=True)
            # 等待任意服务终止
            await asyncio.wait([web_task, self.bot.updater.wait_closed()], return_when=asyncio.FIRST_COMPLETED)


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

关键改动说明 #

• 移除线程与nest_asyncio：放弃用线程隔离Web服务，改用uvicorn.Server的异步serve()方法，将FastAPI纳入主事件循环。
• 统一异步生命周期：Telegram Bot使用async with和异步启动方法，避免启动额外事件循环，与Web服务共享同一循环。
• 并行异步任务：通过asyncio.create_task将Web服务设为后台任务，与Bot运行并行，无需嵌套循环或线程隔离。

额外注意事项 #

• 确保使用python-telegram-bot v20+版本，该版本全面支持异步操作，updater.start_polling()为异步方法。
• 如需优雅停止服务，可添加信号处理逻辑，触发self.bot.stop()和server.should_exit = True终止任务。

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