Как создать несколько параллельных циклов в aiogram?

Question

Исаак

aiohttp , aiogram , python

Как создать несколько параллельных циклов в aiogram?

Как организовать бесконечный цикл для отслеживания состояния ботов, чтобы он работал стабильно, вне зависимости от вызова функции `/check_status`? Я создал функцию `near_player` с бесконечным циклом, который отправляет запрос на сайт каждые 2 секунды для проверки, но сейчас он работает только при вызове `/check_status`. Помимо этого, попытки запустить `async_task` в главной функции не привели к нужному результату.

18 февраля 2024 г. 13:28

Ответы на вопрос 3

Похожие вопросы

Answer 1

Для создания нескольких параллельных циклов в aiogram можно воспользоваться модулем asyncio. Вот пример кода, который демонстрирует создание двух параллельных циклов:

```python
import asyncio
from aiogram import Dispatcher, Bot, types


async def task1():
    while True:
        print("Task 1 is running")
        await asyncio.sleep(1)


async def task2():
    while True:
        print("Task 2 is running")
        await asyncio.sleep(2)


async def main():
    bot = Bot(token="YOUR_BOT_TOKEN")
    dp = Dispatcher(bot)

    task1_handle = asyncio.create_task(task1())
    task2_handle = asyncio.create_task(task2())

    await asyncio.gather(task1_handle, task2_handle)

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

Чтобы организовать бесконечный цикл для отслеживания состояния ботов, можно использовать asyncio.create_task для запуска асинхронной функции в фоновом режиме. Вот пример кода, который демонстрирует запуск бесконечного цикла для проверки статуса ботов каждые 2 секунды:

```python
import asyncio

async def check_status():
    while True:
        print("Checking status...")
        # Здесь можно добавить логику проверки состояния ботов
        await asyncio.sleep(2)

async def main():
    task_handle = asyncio.create_task(check_status())
    await task_handle

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

Вы можете вызвать эту функцию в вашем основном скрипте для запуска бесконечного цикла. Например, вы можете добавить ее в вашу основную функцию, чтобы она запускалась при инициализации бота:

```python
async def near_player():
    while True:
        print("Checking player's status...")
        # Здесь можно добавить логику проверки состояния игрока
        await asyncio.sleep(2)

@dp.message_handler(commands=['check_status'])
async def check_status(message: types.Message):
    await near_player()

async def main():
    bot = Bot(token="YOUR_BOT_TOKEN")
    dp = Dispatcher(bot)

    task_handle = asyncio.create_task(near_player())
    await dp.start_polling()

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

Таким образом, бесконечный цикл будет работать стабильно вне зависимости от вызова функции `/check_status`.

Answer 2

Выносишь цикл проверки в отдельную корутину check_loop(), в main() делаешь 
 async def main():
    logging.basicConfig(level=logging.INFO)
    async with ClientSession() as session:  # сессию лучше по сто раз не создавать
        check_task = asyncio.create_task(check_loop(session))  # создай её однажды и передай в check_loop
        try:
            bot = Bot(
                token='"'
            )
            await dp.start_polling(bot)
        finally:
            check_task.cancel()  # внутри check_loop() сгенерируется прерывание asyncio.CancelledError
 
 Единственное, придётся в коде/конфиге прописать id чата, куда check_loop() будет отправлять сообщение с результатами.

Answer 3

рассказано было уже 100500 раз, надо открыть доку по asyncio и прочитать её, там ты узнаешь про create_task