如何在同一事件循环中运行FastAPI与python-telegram-bot？

我希望FastAPI应用能访问python-telegram-bot的实时bot_data，调用FastAPI接口时，能给bot_data里存储的所有聊天发送消息。

我清楚问题根源：bot.run_polling()和uvicorn.run(...)会启动两个独立的异步循环，必须让它们运行在同一个循环里。

更新1 #

感谢@MatsLindh的建议，我编写了如下函数并在main代码块中调用，但运行不稳定。有时bot.run_polling()能获取正确循环且一切正常，其他时候会因循环不一致报错：

import asyncio
from uvicorn import Config, Server
# --snip--
def run(app: FastAPI, bot:Application):
    # 使用get_event_loop会导致：
    # RuntimeError: Cannot close a running event loop
    # 推测是因为bot.run_polling()调用了loop.run_until_complete()处理不同任务
    # loop = asyncio.get_event_loop()
    loop = asyncio.new_event_loop()
    server = Server(Config(app=app, port=9001))
    loop.create_task(server.serve())

    t = Thread(target=loop.run_forever)
    t.start()

    bot.run_polling()

    t.join()
# --snip--
if __name__ == "__main__":
# --snip--
    run(f_app, bot_app)

我知道可以将bot.run_polling()拆分为多个内部独立调用，但我确信通过这个快捷函数应该可以实现需求。

初始尝试 #

我的简化配置如下：
最初我尝试用multiprocessing.Process而非线程运行，但bot_data始终为空——推测是进程间不共享bot数据，因此必须在同一个进程内运行。但我一直无法将所有组件运行在同一个异步循环中。

# main.py
# python3.10
# pip install fastapi[all] python-telegram-bot
from threading import Thread

import uvicorn
from telegram.ext import Application, ApplicationBuilder, PicklePersistence
from fastapi import FastAPI, Request

BOT_TOKEN = "telegram-bot-token"
MY_CHAT = 123456

class MyApp(FastAPI):
    def add_bot(self, bot_app: Application):
        self.bot_app = bot_app

async def post_init(app: Application):
    app.bot_data["key"] = 42

f_app = MyApp()

@f_app.get("/")
async def test(request: Request):
   app: MyApp = request.app
   bot_app: Application = app.bot_app
   val = bot_app.bot_data.get('key')
   print(f"{val=}")
   await bot_app.bot.send_message(MY_CHAT, f"Should be 42: {val}")


if __name__ == "__main__":
    pers = PicklePersistence("storage")
    bot_app = ApplicationBuilder().token(BOT_TOKEN).post_init(post_init).persistence(pers).build()
    f_app.add_bot(bot_app)

    t1 = Thread(target=uvicorn.run, args=(f_app,), kwargs={"port": 9001})
    t1.start()

    # --- 主线程启动轮询会导致：
    # telegram.error.NetworkError: Unknown error in HTTP implementation:
    # RuntimeError('<asyncio.locks.Event object at 0x7f2764e6fd00 [unset]> is bound to a different event loop')
    # 消息能发送且值正确，但应用崩溃返回500
    # bot_app.run_polling()

    # --- 单独线程启动轮询会导致：
    # RuntimeError: There is no current event loop in thread 'Thread-2 (run_polling)'.
    # t2 = Thread(target=bot_app.run_polling)
    # t2.start()

    # --- 用asyncio启动会导致：
    # ValueError: a coroutine was expected, got <bound method Application.run_polling ...
    # import asyncio
    # t2 = Thread(target=asyncio.run, args=(bot_app.run_polling,))
    # t2.start()

    t1.join()

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