Асинхронна обробка повідомлень від користувача

Якщо ви вже працювали з Python, то, мабуть, чули про ключові слова async і await.

Асинхронне програмування дозволяє нам виконувати задачі "одночасно", навіть якщо вони потребують трохи часу на отримання даних (наприклад, запити до зовнішніх API або взаємодія з базами даних).

Для керування такими задачами Python використовує event loop (цикл подій), який переключається між задачами, коли одна з них "чекає".

Ось простий приклад асинхронного коду:

import asyncio

async def say_hello():
    await asyncio.sleep(2)
    print("Привіт, світе!")

async def main():
    await say_hello()

# Запуск event loop
asyncio.run(main())

Тут функція say_hello "спить" 2 секунди, але блокування основного потоку виконання відсутнє — Python може зайнятися іншими задачами, поки ми, наприклад, чекаємо відповіді від сервера.

У Telegram-ботах це особливо корисно, бо наш бот може одночасно обробляти кілька повідомлень від різних користувачів завдяки асинхронному підходу.

Асинхронність у Telegram-ботах

Тепер давайте застосуємо наші знання асинхронного програмування в розробці Telegram-бота.

Приклад 1: Асинхронне очікування повідомлень

Уявімо, що ваш бот обробляє повідомлення і паралельно має надсилати відповіді. З бібліотекою python-telegram-bot версії 20+ у нас тепер є повноцінна підтримка асинхронного програмування.

from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters

# Асинхронна функція для обробки команди /start
async def start(update: Update, context):
    await update.message.reply_text("Привіт! Я ваш асинхронний бот.")

# Асинхронна функція для обробки текстових повідомлень
async def echo(update: Update, context):
    await update.message.reply_text(f"Ви сказали: {update.message.text}")

if __name__ == "__main__":
    # Створюємо application
    application = Application.builder().token("Ваш_API_токен").build()

    # Регіструємо обробники
    application.add_handler(CommandHandler("start", start))
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))

    # Запуск application
    application.run_polling()

Тут все просто: ми використовуємо ключове слово async для оголошення функцій і await для роботи з асинхронними діями. У нашому випадку це виклики методу reply_text.

Асинхронний запит до зовнішнього API

Ймовірно, ваш бот рано чи пізно захоче поспілкуватися з іншими API. Наприклад, перевірити погоду, перекласти текст або дізнатися курс валют. Ось приклад коду, де бот робить запит випадкового факту до зовнішнього REST API.

Приклад 2: Асинхронна робота з HTTP-запитами

import aiohttp
from telegram import Update
from telegram.ext import Application, CommandHandler

# Асинхронна функція для спілкування з зовнішнім API
async def get_random_fact():
    url = "https://uselessfacts.jsph.pl/random.json?language=en"
    async with aiohttp.ClientSession() as session:  # Створюємо сесію
        async with session.get(url) as response:  # Відправляємо запит
            data = await response.json()  # Асинхронно читаємо JSON
            return data.get("text", "Не вдалося отримати факт.")

# Асинхронна команда /fact
async def fact(update: Update, context):
    fact_text = await get_random_fact()  # Чекаємо результат від API
    await update.message.reply_text(f"Випадковий факт: {fact_text}")

if __name__ == "__main__":
    application = Application.builder().token("Ваш_API_токен").build()

    application.add_handler(CommandHandler("fact", fact))

    application.run_polling()

У цьому прикладі ми використовуємо бібліотеку aiohttp для відправки асинхронного HTTP-запиту. Коли користувач надсилає /fact, бот запитує зовнішній API, чекає відповідь і повертає користувачу випадковий факт.

Покращення продуктивності бота

Уявіть, що ваш бот має трохи зачекати перед відправкою повідомлення (наприклад, для ефекту "набирання тексту"). Асинхронність дозволяє зробити це елегантно:

import asyncio

async def typing_effect(update: Update, context):
    await update.message.reply_text("Почекайте, я думаю...")
    await asyncio.sleep(2)  # Затримка
    await update.message.reply_text("Відповідь готова!")

Простий, але ефективний спосіб показати користувачу, що бот "працює".

Часті помилки та як їх уникнути

Асинхронне програмування — це круто, але іноді воно може стати джерелом головного болю. Наприклад, якщо ви забудете додати await, код може просто не виконатися. Погодьтесь, дивно, коли ваш бот мовчить, хоча має щось робити.

Інша типова проблема — неправильна обробка помилок в асинхронних функціях. Якщо зовнішнє API недоступне або повертає помилку, ви можете отримати негарний traceback. Краще завжди використовувати блок try-except в асинхронних викликах:

async def safe_request():
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get("https://example.com/api") as response:
                return await response.json()
    except Exception as e:
        print(f"Помилка запиту: {e}")

Інтеграція з вебхуками

Тепер давайте поєднаємо все, що ми вивчили, за допомогою вебхуків, щоб наш бот міг обробляти повідомлення в реальному часі. У цьому випадку FastAPI буде виступати в ролі "моста", який обробляє вхідні запити від Telegram.

Приклад 3: Асинхронний бот з вебхуками

from fastapi import FastAPI, Request
from telegram import Update
from telegram.ext import Application, CommandHandler

app = FastAPI()

application = Application.builder().token("Ваш_API_токен").build()

@app.post("/")
async def handle_webhook(request: Request):
    json_data = await request.json()
    update = Update.de_json(json_data, application.bot)
    await application.process_update(update)

async def start(update: Update, context):
    await update.message.reply_text("Привіт! Цей бот працює через вебхуки.")

application.add_handler(CommandHandler("start", start))

if __name__ == "__main__":
    import uvicorn
    # Вказуємо webhook URL (наприклад, https://yourserver.com/)
    application.bot.set_webhook("https://ваш-домен/")
    uvicorn.run(app, host="0.0.0.0", port=8000)

Тепер Telegram надсилає оновлення на ваш сервер, який обробляє їх за допомогою FastAPI.

Практичне застосування

Створення асинхронних ботів дозволяє покращити продуктивність застосунку і одночасно обробляти запити від великої кількості користувачів. Асинхронна обробка даних затребувана на співбесідах і допомагає писати масштабований код у реальних проєктах. Ваші знання знадобляться для створення складних систем, таких як чат-боти для техпідтримки, маркетингові інструменти і навіть ігрові платформи.