Примеры решений для Python бота

Приветствую! Здесь вы наверняка найдете, что ищете. Примеры в лаборатории рассчитаны на то, что мы разбираем что-то конкретное.

Текущая статья посвящена bot, диалоги, БД, безопасность и деплой..

Поэтому за теорией по текущей теме вам — в энциклопедию. Если ещё не погружались, то маршрут прост:

1. Основы
2. Система и сеть
3. Данные и разметка
4. Код и разработка
5. Языки
6. Искусственный интеллект
7. Проект
8. Инфраструктура и безопасность
9. Спин-офф

Обязательно пройдитесь.

А теперь приступим к нашему предмету.

---

---

Архитектура проекта

bot/
  main.py           # точка входа
  settings.py       # конфиг из env
  handlers/
    start.py
    survey.py
  services/
    user_service.py
  repositories/
    user_repo.py

Слой	Ответственность
handlers	Команды, кнопки, текст пользователя
services	Бизнес-правила
repositories	SQL/ORM, Redis

---

Шаг 1. Минимальный запуск (PTB v20+)

Проверка: бот отвечает на /start в Telegram.

---

Шаг 2. Диалог с валидацией (ConversationHandler)

1. Пользователь вводит /survey.
2. Бот спрашивает имя → проверяет длину.
3. Спрашивает возраст → проверяет число.
4. Сохраняет в БД и отправляет итог.

async def ask_age(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    text = (update.message.text or "").strip()
    if not text.isdigit() or not (1 <= int(text) <= 120):
        await update.message.reply_text("Введите возраст числом от 1 до 120")
        return ASK_AGE  # константа состояния из ConversationHandler
    context.user_data["age"] = int(text)
    await update.message.reply_text("Спасибо, данные сохранены")
    return ConversationHandler.END  # завершение диалога

---

Шаг 3. Inline-кнопки

from telegram import InlineKeyboardButton, InlineKeyboardMarkup

keyboard = [[InlineKeyboardButton("Да", callback_data="yes"),
             InlineKeyboardButton("Нет", callback_data="no")]]
await update.message.reply_text("Продолжить?", reply_markup=InlineKeyboardMarkup(keyboard))

---

Шаг 4. Интеграция с API

import httpx

async def fetch_rate() -> float:
    async with httpx.AsyncClient(timeout=10.0) as client:
        r = await client.get("https://api.example.com/rate")
        r.raise_for_status()
        return r.json()["value"]

Добавьте retry и кэш для внешних сервисов.

---

Шаг 5. База данных (asyncpg)

async with pool.acquire() as conn:
    await conn.execute(
        "INSERT INTO users (tg_id, name) VALUES ($1, $2) ON CONFLICT (tg_id) DO UPDATE SET name = $2",
        tg_id, name,
    )

---

Безопасность и эксплуатация

• Rate limit на тяжёлые команды (Redis + счётчик).
• Валидация и экранирование пользовательского ввода.
• Webhook только по HTTPS с проверкой secret token.
• Health endpoint и метрики (Prometheus) при деплое в Docker.
• Логи в JSON, correlation id на запрос.

---

Деплой

1. Docker-образ с python:3.12-slim.
2. Переменные BOT_TOKEN, DATABASE_URL.
3. Webhook: application.run_webhook(..) за reverse proxy.
4. Systemd или Kubernetes для перезапуска при падении.

---

Чек-лист готовности

• Бот переживает рестарт без потери критичных данных.
• Ошибки не роняют процесс (глобальный error handler).
• Админ-команды защищены списком ADMIN_IDS.
• Есть тесты на handlers (pytest + моки Update).