相关疑难解决方法(0)

我正在使用 FastAPI 在 python 中创建一个服务器，并且我想要一个与我的 API 无关的函数，每 5 分钟在后台运行一次（例如从 API 检查内容并根据响应打印内容）

我尝试创建一个运行该函数的线程start_worker，但它不打印任何内容。

有谁知道该怎么做？

def start_worker():
    print('[main]: starting worker...')
    my_worker = worker.Worker()
    my_worker.working_loop() # this function prints "hello" every 5 seconds

if __name__ == '__main__':
    print('[main]: starting...')
    uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)
    _worker_thread = Thread(target=start_worker, daemon=False)
    _worker_thread.start()

我希望我的 FastAPI 应用程序能够访问始终实际bot_data的python-telegram-bot。我需要这个，所以当我调用 FastAPI 中的某个端点时，可以将消息发送到所有聊天，存储在bot_data.

据我了解这个问题：bot.run_polling()并uvicorn.run(...)启动两个独立的异步循环。我需要将它们整合在一起运行。

UPD-1：
感谢@MatsLindh，我创建了下一个函数，我将其传递给主块，但它的工作不一致。有时bot.run_polling()（获得正确的循环并且一切正常，但其他时候并因存在不同循环的错误而中断）：

import asyncio
from uvicorn import Config, Server
# --snip--
def run(app: FastAPI, bot:Application):
    # using get_event_loop leads to:
    # RuntimeError: Cannot close a running event loop
    # I guess it is because bot.run_polling()
    # calls loop.run_until_complete() different tasks
    # 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() …