Как запустить aiogram webhook и aio-pika в одном event loop?

Question

[milltt]

Есть вот такой consumer из оф. документации aio-pika:

async def orders_queue(): 
     connection = await aio_pika.connect_robust( 
         host="localhost" 
     ) 
  
     queue_name = "orders" 
     async with connection: 
         channel = await connection.channel() 
  
         await channel.set_qos(prefetch_count=10) 
         queue = await channel.declare_queue(queue_name) 
  
         async with queue.iterator() as queue_iter: 
             async for message in queue_iter: 
                 async with message.process(): 
                     order_id = message.body.decode() 
                     print(order_id) 
  
                     if queue.name in message.body.decode(): 
                         break

На стороне крутится FastApi сервис, который принимает callback от эквайринга и обновляет статус заказа в бд. Необходимо реализовать создание заказа на стороне другого агрегатора непосредственно в боте, так как при проектировании библиотеки, модели и схемы данных не разнесли по отдельным сервисам, а переносить это все в сервис для работы с эквайрингом не целесообразно. Я решил, что можно с этого сервиса публиковать сообщение в очередь и обрабатывать его на стороне бота, но если запускать бота с помощью вебхука, а эту функцию-consumer с помощью create_task, то айограммовский вебхук блокирует ее дальнейшее выполнение, бот работает, но сообщения из очереди кролика больше не принимает. Есть ли вариант обхода блокировки loop'а стартером вебхука, чтобы и бот продолжал принимать апдейты и consumer не блокировался?

Comments

[milltt]

Я ещё не пробовал использовать потоки, помимо create_task и прочего, если я передам сюда текущий loop, это поможет запустить вебхук и aio-pika параллельно? asyncio.run_coroutine_threadsafe()

(прямо сейчас не могу проверить, будет здорово, если у кого то есть ответ на этот новый вопрос)

[Модератор]

Не надо ставить как можно больше тэгов. Лучше оставить один, но конкретный, с которым проблема.
См.п.3.1 Регламента.

[milltt]

Justa Gain, нашел вот такое решение, не знаю, насколько оно верное, однако получилось подружить RabbitMQ и Aiogram в одном runtime. Спасибо за наводку на то, что в on_startup можно прописать что-то помимо установки вебхука.

def start_background_loop(side_loop: asyncio.AbstractEventLoop) -> None:
    asyncio.set_event_loop(side_loop)
    side_loop.run_forever()

async def on_startup(dp):
    await bot.set_webhook(config.tg_bot.webhook_url)

    new_loop = asyncio.new_event_loop()
    t = Thread(target=start_background_loop, args=(new_loop,), daemon=True)
    t.start()

    asyncio.run_coroutine_threadsafe(orders_queue(), new_loop)

if __name__ == "__main__":
        loop.create_task(main(dp, config))

        executor.start_webhook(
            dp,
            config.tg_bot.webhook_path,
            on_startup=on_startup,
            on_shutdown=on_shutdown,
            skip_updates=True
        )

[milltt]

Justa Gain, я только вчера пробовал через create_task запускать эту функцию, как с помощью loop'а, так и напрямую, в том числе и в теле on_startup. Только сейчас заработало, честно, не знаю почему, спасибо!

[milltt]

Justa Gain, прошу прощения, создавал таск не в теле on_startup, видимо по этой причине и не работало, еще раз спасибо большое!

Answer 1

[milltt]

Запускать с помощью asyncio.create_task в теле функции on_startup

async def on_startup(dp):
    await bot.set_webhook(config.tg_bot.webhook_url)

    asyncio.create_task(orders_queue())