Aiogram — это современный асинхронный фреймворк для создания Telegram-ботов на Python. В отличие от синхронных библиотек, aiogram построен на основе asyncio, что позволяет обрабатывать множество запросов одновременно без блокировок. Это делает ботов более отзывчивыми и производительными.
Основные преимущества aiogram
- Полная асинхронность
- Простота использования
- Поддержка всех возможностей Telegram Bot API
- Гибкая система middleware
- Регулярные обновления
Установка и настройка
Для начала работы установите aiogram через pip
pip install aiogram
Создайте нового бота через [BotFather](httpst.meBotFather) и получите API-токен.
Базовая структура бота
Создайте файл `bot.py` со следующим содержимым
from aiogram import Bot, Dispatcher, executor, types
# Настройка логирования
logging.basicConfig(level=logging.INFO)
# Инициализация бота и диспетчера
API_TOKEN = 'YOUR_API_TOKEN'
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)
# Обработчик команды start
@dp.message_handler(commands=['start'])
async def send_welcome(message types.Message)
....await message.reply(Привет! Я эхо-бот на aiogram.)
# Обработчик текстовых сообщений
@dp.message_handler()
async def echo(message types.Message)
....await message.answer(message.text)
# Запуск бота
if __name__ == '__main__'
....executor.start_polling(dp, skip_updates=True)
Этот код создает простого эхо-бота, который отвечает на любое сообщение тем же текстом.
Обработчики сообщений
Фильтры команд
@dp.message_handler(commands=['start', 'help'])
async def send_welcome(message types.Message)
....await message.answer(Добро пожаловать! Доступные команды start, help, info)
Фильтры по содержанию
@dp.message_handler(content_types=types.ContentType.TEXT)
async def handle_text(message types.Message)
....await message.answer(Вы отправили текстовое сообщение!)
@dp.message_handler(content_types=types.ContentType.PHOTO)
async def handle_photo(message types.Message)
....await message.answer(Вы отправили фото!)
Кастомные фильтры
from aiogram import filters
@dp.message_handler(filters.Text(equals='привет', ignore_case=True))
async def hello_handler(message types.Message)
....await message.answer(И тебе привет!)
@dp.message_handler(filters.Text(contains='python', ignore_case=True))
async def python_handler(message types.Message)
....await message.answer(Вы упомянули Python!)
Клавиатуры и кнопки
Reply-клавиатура
from aiogram.types import ReplyKeyboardMarkup, KeyboardButton
# Создание клавиатуры
kb = ReplyKeyboardMarkup(
resize_keyboard=True, # автоматическое изменение размера
one_time_keyboard=True # скрыть после использования
)
# Добавление кнопок
btn1 = KeyboardButton('Кнопка 1')
btn2 = KeyboardButton('Кнопка 2')
btn3 = KeyboardButton('Кнопка 3')
kb.add(btn1, btn2).add(btn3)
@dp.message_handler(commands=['start'])
async def send_welcome(message types.Message)
....await message.answer(Выберите действие, reply_markup=kb)
@dp.message_handler(filters.Text(equals='Кнопка 1'))
async def btn1_handler(message types.Message)
....await message.answer(Вы нажали кнопку 1!)
Inline-клавиатура
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
# Создание inline-клавиатуры
inline_kb = InlineKeyboardMarkup()
btn1 = InlineKeyboardButton('Кнопка 1', callback_data='btn1')
btn2 = InlineKeyboardButton('Кнопка 2', callback_data='btn2')
inline_kb.add(btn1, btn2)
@dp.message_handler(commands=['inline'])
async def show_inline(message types.Message)
....await message.answer(Выберите действие, reply_markup=inline_kb)
# Обработчик нажатий на inline-кнопки
@dp.callback_query_handler(lambda c c.data == 'btn1')
async def process_callback_btn1(callback_query types.CallbackQuery)
....await bot.answer_callback_query(callback_query.id)
....await bot.send_message(callback_query.from_user.id, 'Вы нажали кнопку 1!')
Finite State Machine (FSM)
FSM позволяет управлять состоянием диалога с пользователем.
from aiogram.contrib.fsm_storage.memory import MemoryStorage
from aiogram.dispatcher import FSMContext
from aiogram.dispatcher.filters.state import State, StatesGroup
# Инициализация хранилища состояний
storage = MemoryStorage()
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot, storage=storage)
# Определение состояний
class Form(StatesGroup)
....name = State()
....age = State()
....gender = State()
# Начало диалога
@dp.message_handler(commands='form')
async def cmd_start(message types.Message)
....await Form.name.set()
....await message.reply(Как вас зовут)
# Обработка имени
@dp.message_handler(state=Form.name)
async def process_name(message types.Message, state FSMContext):
....async with state.proxy() as data:
........data['name'] = message.text
....await Form.next()
....await message.reply(Сколько вам лет)
# Обработка возраста
@dp.message_handler(state=Form.age)
async def process_age(message types.Message, state FSMContext):
....if not message.text.isdigit():
........await message.reply(Пожалуйста, введите число!)
........return
....async with state.proxy() as data:
........data['age'] = int(message.text)
....await Form.next()
....await message.reply(Какой ваш пол)
# Обработка пола и завершение
@dp.message_handler(state=Form.gender)
async def process_gender(message types.Message, state FSMContext):
....async with state.proxy() as data:
........data['gender'] = message.text
....# Завершаем состояние
....await state.finish()
....# Отправляем собранные данные
....await message.reply(
........f'Спасибо! Ваши данные\n'
........f'Имя {data['name']}\n'
........f'Возраст {data['age']}\n'
........f'Пол {data['gender']}'
....)
Работа с медиафайлами
Отправка фото
@dp.message_handler(commands=['photo'])
async def send_photo(message types.Message):
....# Отправка фото по URL
....await bot.send_photo(
........chat_id=message.chat.id,
........photo=httpsexample.comimage.jpg,
........caption=Это пример фото
....)
....# Отправка фото из файла
....with open('local_image.jpg', 'rb') as photo:
........await bot.send_photo(
............chat_id=message.chat.id,
............photo=photo,
............caption=Локальное фото
........)
Загрузка полученных файлов
@dp.message_handler(content_types=types.ContentType.PHOTO)
async def download_photo(message types.Message)
....# Получаем информацию о файле
....file_id = message.photo[-1].file_id
....file = await bot.get_file(file_id)
....file_path = file.file_path
....# Скачиваем файл
....await bot.download_file(file_path, 'downloaded_photo.jpg')
....await message.answer(Фото сохранено!)
Middleware и кастомизация
Создание middleware
class CounterMiddleware
....def __init__(self)
........self.counter = 0
....async def on_process_message(self, message types.Message, data dict)
........self.counter += 1
........logging.info(f'Сообщение #{self.counter}')
# Регистрация middleware
dp.middleware.setup(CounterMiddleware())
Логирование
import logging
from aiogram import types
from aiogram.dispatcher.handler import CancelHandler
from aiogram.dispatcher.middlewares import BaseMiddleware
class LoggingMiddleware(BaseMiddleware)
....async def on_pre_process_update(self, update types.Update, data dict)
........logging.info(fПолучено обновление {update})
....async def on_pre_process_message(self, message types.Message, data dict)
........logging.info(fПолучено сообщение от {message.from_user.id} {message.text})
........# Пример фильтрации
........if спам in message.text.lower()
............logging.warning(Обнаружено спам-сообщение!)
............await message.answer(Сообщение содержит спам!)
............raise CancelHandler() # Отменяем дальнейшую обработку
Webhook вместо Long Polling
Для production-среды лучше использовать webhook
from aiogram import Bot, Dispatcher, types
from aiogram.contrib.fsm_storage.memory import MemoryStorage
from aiohttp import web
API_TOKEN = 'YOUR_API_TOKEN'
WEBHOOK_HOST = 'httpsyour.domain'
WEBHOOK_PATH = 'webhook'
WEBHOOK_URL = f'{WEBHOOK_HOST}{WEBHOOK_PATH}'
# Инициализация
bot = Bot(token=API_TOKEN)
storage = MemoryStorage()
dp = Dispatcher(bot, storage=storage)
# Обработчики
@dp.message_handler(commands=['start'])
async def send_welcome(message types.Message)
....await message.answer(Привет! Я бот на webhook!)
....# Настройка webhook
async def on_startup(app):
....await bot.set_webhook(WEBHOOK_URL)
async def on_shutdown(app)
....await bot.delete_webhook()
# Создание aiohttp приложения
app = web.Application()
app.router.add_post(WEBHOOK_PATH, dp._process_update)
app.on_startup.append(on_startup)
app.on_shutdown.append(on_shutdown)
if __name__ == '__main__'
....web.run_app(app, host='0.0.0.0', port=3000)
Пример полноценного бота
Создаем бота-опросника
import logging
from aiogram import Bot, Dispatcher, executor, types
from aiogram.contrib.fsm_storage.memory import MemoryStorage
from aiogram.dispatcher import FSMContext
from aiogram.dispatcher.filters.state import State, StatesGroup
logging.basicConfig(level=logging.INFO)
API_TOKEN = 'YOUR_API_TOKEN'
bot = Bot(token=API_TOKEN)
storage = MemoryStorage()
dp = Dispatcher(bot, storage=storage)
# Состояния для опроса
class PollStates(StatesGroup)
....question = State()
....options = State()
# Хранение опросов
polls = {}
# Команда начала создания опроса
@dp.message_handler(commands=['create_poll'])
async def cmd_create_poll(message types.Message)
....await PollStates.question.set()
....await message.answer(Введите вопрос для опроса)
# Получение вопроса
@dp.message_handler(state=PollStates.question)
async def process_question(message types.Message, state FSMContext)
....async with state.proxy() as data:
........data['question'] = message.text
........data['creator'] = message.from_user.id
....await PollStates.next()
....await message.answer(Теперь введите варианты ответов через запятую)
# Получение вариантов ответов
@dp.message_handler(state=PollStates.options)
async def process_options(message types.Message, state FSMContext)
....options = [opt.strip() for opt in message.text.split(',') if opt.strip()]
....if len(options) < 2:
........await message.answer(Нужно как минимум два варианта ответа!)
........return
....async with state.proxy() as data:
........poll_id = str(message.message_id)
........polls[poll_id] = {
............'question' data['question'],
............'options' options,
............'votes' {opt 0 for opt in options},
............'voted' set()
........}
....# Создание inline-клавиатуры с вариантами
....keyboard = types.InlineKeyboardMarkup()
....for option in options
........keyboard.add(types.InlineKeyboardButton(
............text=option,
............callback_data=fvote_{poll_id}_{option}
........))
....await state.finish()
....await message.answer(fОпрос {data['question']}, reply_markup=keyboard)
# Обработка голосов
@dp.callback_query_handler(lambda c c.data.startswith('vote_'))
async def process_vote(callback_query types.CallbackQuery):
...._, poll_id, option = callback_query.data.split('_', 2)
....if poll_id not in polls:
........await callback_query.answer(Опрос не найден!)
........return
....poll = polls[poll_id]
....if callback_query.from_user.id in poll['voted']:
........await callback_query.answer(Вы уже голосовали в этом опросе!)
........return
....# Обновление результатов
....poll['votes'][option] += 1
....poll['voted'].add(callback_query.from_user.id)
....# Обновление сообщения с результатами
....results = n.join([f{opt} {count} for opt, count in poll['votes'].items()])
....await callback_query.message.edit_text(
........f'{poll['question']}nnРезультатыn{results}'
....)
....await callback_query.answer(fВы проголосовали за {option})
if __name__ == '__main__'
....executor.start_polling(dp, skip_updates=True)
Лучшие практики и советы
1. Обработка ошибок Всегда обрабатывайте возможные исключения
2. Валидация данных Проверяйте ввод пользователя
3. Кэширование Используйте кэш для часто запрашиваемых данных
4. База данных Для production-ботов используйте базу данных вместо MemoryStorage
5. Безопасность Никогда не храните токен в коде, используйте переменные окружения
6. Логирование Настройте подробное логирование для отладки
Заключение
Aiogram — это мощный и гибкий фреймворк для создания Telegram-ботов на Python. Он предоставляет все необходимые инструменты для создания как простых, так и сложных ботов с богатой функциональностью. Асинхронная природа библиотеки обеспечивает высокую производительность, а понятный API делает разработку приятной и эффективной.
Это руководство покрывает основные аспекты работы с aiogram, но библиотека предлагает еще больше возможностей работу с платежами, глубокую интеграцию с Telegram API, кастомные фильтры и многое другое.
Подписывайтесь:
Телеграм https://t.me/lets_go_code
Канал "Просто о программировании" https://dzen.ru/lets_go_code