В сети куча туториалов по синхронным ботам и почти ничего по асинхронным. Статья нацелена на новичков в асинхронном программировании в целом и в асинхронных ботах в частности. В этой статье не будет глубокого анализа асинхронности и технических деталей реализации со сложными терминами, только суть и практические примеры.

Тезаурус. Ссылки

Здесь приведено пояснение основных терминов асинхронного программирования, которые использованы в статье. Это сделано для того, чтобы не объяснять некоторые вещи на ходу.

• Корутина (сопрограмма) — функция, объявленная с помощью async def func(). Может приостанавливать свое выполнение с благодаря await. Возвращает футуру.

• Футура — объект-контейнер для результата выполнения функции, которого еще нет. Получают с помощью await func().

• Задача — обертка над корутиной, запускаемая с помощью asyncio.run(), asyncio.create_task(), asyncio.gather() и т. д. Задачи запускаются в так называемом цикле событий (event loop), которым управляет модуль asyncio. Задача — ключевая сущность, позволяющая запускать корутины "паралелльно".

• asyncio — стандартная библиотека Python, отвечающая за запуск асинхронных функций и осуществляющая управление циклом событий.

• I/O-bound (от англ. Input/Output-bound) — задачи, ограниченные вводом-выводом. Например, чтение/запись файлов, получение/отправка сетевых запросов.

• CPU-bound — задачи, ограниченные производительностью процессора. Например, работа с большими объемами данных, рендеринг графики, криптография.

Полезные ссылки

• PyTelegramBotAPI

• asyncio

• aiohttp

• aiofiles

• Цикл лекций от Яндекса на YouTube

Суть асинхронности

Синхронный код выполняется строка за строкой, и в этом его прелесть — его намного проще читать и понимать. Асинхронный код, как следует из названия, выполняется не последовательно, из-за чего большинство новичков обходят эту тему стороной и боятся как огня. Контент этого параграфа нацелен на простое, метафорическое объяснение асинхронности. Идея взята из документации FastAPI.

Представьте себе, что вы находитесь в бургерной и хотите сделать большой заказ. Вы точно уверены в том, что приготовление вашего заказа — продолжительная по времени операция. Вы уже стоите на кассе и заказывает большой комбо с биг тейсти, картошкой по-деревенски, колой и, возможно, морковными палочками. Кассир принимает ваш заказ, отдает чек и вы садитесь за свой столик. Вы не можете сразу получить свой заказ, поэтому приостанавливаете выполнение функции получение заказа на моменте заказ оформлен.

Теперь вам остается только ждать в предвкушении. Так как большое комбо делается не за 2 минуты, то вы наверняка решите занятся чем-то полезным и продуктивным, например, поскроллить ленту почитать книгу. Вы переключаете свое внимание на книгу и начинаете выполнять другую функцию — чтение книги, пока готовится ваш заказ (выполняется функция получение заказа).

Периодически вы поглядываете на табло в поиске своего номера, продолжая читать книгу. Спустя 10 минут наконец-то ваш заказ готов и вы идете на кассу его получать, завершая выполнение функции чтение книги. Вы продолжили выполнять функцию получение заказа с момента остановки (заказ оформлен). Вы забрали большой комбо, вернулись за столик и приступили к наслаждению пищей.

Если бы эта маленькая история была программой на Python, то она бы выглядела так:

import asyncio
import random

async def make_burger(order):
    """Готовим бургер (долгая операция)"""
    print(f"? Начали готовить {order}")
    cooking_time = random.randint(3, 8)
    await asyncio.sleep(cooking_time)  # Ждём, пока приготовится
    print(f"✅ {order} готов! (заняло {cooking_time} сек)")
    return f"Вкусный {order}"

async def read_book():
    """Читаем книгу, пока ждём заказ"""
    for page in range(1, 4):
        print(f"? Читаю страницу {page}...")
        await asyncio.sleep(2)  # Читаем по 2 секунды на страницу

async def visit_burger_shop():
    """Посещение бургерной (главная функция)"""
    print("? Я зашёл в бургерную")
    
    # Делаем заказ и начинаем готовить
    burger_task = asyncio.create_task(make_burger("Биг Тейсти"))
    
    # Пока готовят - читаем книгу
    print("\nПока жду заказ, почитаю книгу:")
    reading_task = asyncio.create_task(read_book())
    
    # Ждём, пока бургер будет готов
    await burger_task
    
    # Перестаём читать, когда еда готова
    reading_task.cancel()
    print("\n? Закрываю книгу - заказ готов!")
    
    # Получаем заказ
    print(f"\n? Получил {burger_task.result()}! Можно кушать.")

# Запускаем
asyncio.run(visit_burger_shop())

Результат выполнения кода примерно такой:

? Я зашёл в бургерную

Пока жду заказ, почитаю книгу:
? Начали готовить Биг Тейсти
? Читаю страницу 1...
? Читаю страницу 2...
✅ Биг Тейсти готов! (заняло 3 сек)

? Закрываю книгу - заказ готов!

? Получил Вкусный Биг Тейсти! Можно кушать.

Мы не ждали, пока приготовится бургер — мы отложили эту задачу и продолжили выполнять полезную работу (читать). Как только бургер приготовился, мы продолжили выполнение изначальной функции (получение заказа и употребление бургера в пищу). В этом суть асинхронности.

Где уместно использовать асинхронность

Асинхронность позволяет не ждать выполнения долгих операций и минимизировать время простоя процессора. Процессор выполняет математические операции, но зачастую ему приходится ждать получения данных для вычислений. Получается, что выполнение такой программы будет ограничено временем чтения/записи данных с диска или получением данных из сетевого запроса. Такие операции называются I/O-bound (от англ. input/output bound). Так вот асинхронность эффективно решает именно такие задачи, где требуется ждать данные. Если в вашем боте (или любой другой программе) требуется выполнять много сетевых запросов, например к внешнему API, или читать данные с диска, то лучшим выбором для реализации программы, вероятно, будет асинхронность.

Где неуместна асинхронность

Помимо I/O-bound задач существуют также CPU-bound, по аналогии — ограниченные вычислительной мощностью процессора. Типичные сценарии: машинное обучение, криптография, рендеринг графики, обработка больших объемов данных. Такие задачи решаются преимущественно многопоточностью и многопроцессорностью, но это совсем другая история. Использование асинхронности в таких задачах не только усложняет и без того сложный код, но и потенциально замедляет работу программы.

Асинхронность — великая сила. С велий силой приходит великая ответственность.

PyTelegramBotAPI

Установка

Если вы еще не установили библиотеку, то прямо сейчас самое время:

pip install pytelegrambotapi

Эхо-бот

Пример простого эхо-бота (далее пример реального бота):

import asyncio

from telebot.async_telebot import AsyncTeleBot

bot = AsyncTeleBot('TOKEN')

# Обрабатывает команды /help и /start
@bot.message_handler(commands=['help', 'start'])
async def send_welcome(message):
    text = 'Привет, я эхо-бот. Напиши мне что-нибудь и я повторю это!'
    await bot.reply_to(message, text)


# Обрабатывает все входящие текстовые сообщения. По-умолчанию content_types=['text']
@bot.message_handler(func=lambda message: True)
async def echo_message(message):
    await bot.reply_to(message, message.text)


asyncio.run(bot.infinity_polling())

Сначала импортируем необходимые библиотеки asyncio и pytelegrambotapi (telebot). Класс синхронного бота лежит в главном файле, то есть если бы мы писали синхронный бот, то вместо from telebot.async_telebot import AsyncTeleBot мы бы написали from telebot import TeleBot.

Пишем два обработчика: на команды (send_welcome) и на текстовые сообщения (echo_message). Обратите внимание, что в объявлении функций использовалось async, тобишь мы создали корутину.

Основная функция для запуска бота с помощью long polling — bot.infinity_polling(). Так как все методы класса AsyncTeleBot асинхронные, то мы используем asyncio.run(), чтобы запустить функцию.

Если вы уже знакомы с синхронной реализацией библиотеки, то вам не составит труда реализовать все в асинхронном виде, т. к. методы AsyncTeleBot и TeleBot абсолютно одинаковые.

Пример реального бота

У меня есть бот, который парсит музыкальный сайт и извлекает оттуда треки по запросу. Естественно, я использую его исключительно в личных целях. На большинство запросов приходится порядка 300-500 результатов, которые извлекаются сразу с нескольких страниц сайта (иногда 100-150 одновременно), поэтому наиболее эффективным решением была именно асинхронная реализация.

Фрагмент кода:

import asyncio, aiohttp, aiofiles

from telebot.async_telebot import AsyncTeleBot


bot = AsyncTeleBot('TOKEN', 'html')


@bot.message_handler(commands=['start'])
async def start(m: M):
    text = 'Пришли название трека, артиста, альбома'
    await bot.reply_to(m, text)


@bot.message_handler()
async def search(m: M):
    async with MusicParser() as parser:
        result = await parser.search_tracks(m.text)
    
    if not result:
        await bot.reply_to(m, f'По запросу <b>{m.text}</b> ничего не найдено')
        return
    
    tracks = result['tracks']
    albums = result['albums']
    collections = result['collections']
    artists = result['artists']
    radio = result['radio']

    await bot.reply_to(m, f'Найдено {len(tracks)} треков, {len(albums)} альбомов, {len(artists)} исполнителей и {len(collections)} коллекций по запросу <b>{m.text}</b>')


async def download_track(url: str):
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                if response.status == 200:
                    filename = f"mp3/{url.split('/')[-1].split('?')[0]}"
                    async with aiofiles.open(filename, 'wb') as f:
                        await f.write(await response.read())
    except Exception as e:
        print(f"Ошибка загрузки {url}: {e}")


async def main():
    print('Бот запущен!')
    await bot.infinity_polling(60, True)


if __name__ == '__main__':
    asyncio.run(main())

Повторяться не будем, расскажу по делу. Функция search() получает результат по запросу с помощью некоего класса MusicParser и выводит количество результатов по каждой сущности.

Функция download_track() получает ссылку на скачивание и с помощью асинхронной библиотеки aiohttp скачивает трек. Далее асинхронная библиотека aiofiles записывает скачанные данные на диск в папку mp3. aiofiles.open() фактически асинхронный аналог встроенной функции Python open(). Сетевые запросы и запись данных на диск используются в блоке контекстного менеджера with, чтобы автоматически освобождать ресурсы после выполненной операции.

Вместо заключения

Если вы не поняли асинхронность, прочитав массу статей и просмотрев сотню примеров, то знайте, что это нормально. Пока самостоятельно не столкнешься с необходимостью ускорения программы, не поймешь, что к чему и как работает. Я считаю, что лучшая тренировка — реальный бой. А еще помогает в понимании кода простое удаление async/await.

Спасибо, что прочитали статью до конца! ?