Библиотека Aiogram: создание Telegram ботов с асинхронностью Python

Расширенные примеры использования Aiogram 3.x

Пример 1: Middleware для измерения времени обработки

import time
from aiogram import BaseMiddleware
from typing import Callable, Dict, Any

class TimingMiddleware(BaseMiddleware):
    async def __call__(self, handler: Callable, event: Any, data: Dict[str, Any]) -> Any:
        start = time.perf_counter()
        result = await handler(event, data)
        duration = time.perf_counter() - start
        print(f'Обработка {event.__class__.__name__} заняла {duration:.3f} сек.')
        return result

dp.update.middleware(TimingMiddleware())

Вывод в консоль при получении сообщения:
Обработка Message заняла 0.021 сек.

Пример 2: FSM с валидацией и отменой

from aiogram.fsm.state import StatesGroup, State
from aiogram.fsm.context import FSMContext
from aiogram.types import Message
from aiogram.filters import Command

class Survey(StatesGroup):
    name = State()
    age = State()
    city = State()

@router.message(Command('survey'))
async def start_survey(message: Message, state: FSMContext):
    await state.set_state(Survey.name)
    await message.answer('Как вас зовут?')

@router.message(Survey.name)
async def process_name(message: Message, state: FSMContext):
    if len(message.text) < 2:
        await message.answer('Имя должно содержать хотя бы 2 символа. Попробуйте ещё раз.')
        return
    await state.update_data(name=message.text)
    await state.set_state(Survey.age)
    await message.answer('Сколько вам лет?')

@router.message(Survey.age)
async def process_age(message: Message, state: FSMContext):
    if not message.text.isdigit():
        await message.answer('Возраст должен быть числом. Введите корректно.')
        return
    age = int(message.text)
    if age < 1 or age > 150:
        await message.answer('Некорректный возраст (1-150).')
        return
    await state.update_data(age=age)
    await state.set_state(Survey.city)
    await message.answer('Из какого вы города?')

@router.message(Survey.city)
async def process_city(message: Message, state: FSMContext):
    data = await state.get_data()
    await message.answer(f'Спасибо за участие! Ваши данные:\nИмя: {data["name"]}\nВозраст: {data["age"]}\nГород: {message.text}')
    await state.clear()

# Хендлер для отмены
@router.message(Command('cancel'))
async def cancel(message: Message, state: FSMContext):
    current_state = await state.get_state()
    if current_state is None:
        return
    await state.clear()
    await message.answer('Диалог отменён.')

Диалог:
Пользователь: /survey
Бот: Как вас зовут?
Пользователь: А
Бот: Имя должно содержать хотя бы 2 символа. Попробуйте ещё раз.
Пользователь: Анна
Бот: Сколько вам лет?
Пользователь: 25
Бот: Из какого вы города?
Пользователь: Москва
Бот: Спасибо за участие! Ваши данные: ...

Пример 3: Отправка альбома (MediaGroup)

from aiogram.types import MediaGroup, InputMediaPhoto

@router.message(Command('album'))
async def send_album(message: Message):
    album = MediaGroup(
        media=[
            InputMediaPhoto(media='https://picsum.photos/200/300?random=1'),
            InputMediaPhoto(media='https://picsum.photos/200/300?random=2', caption='Второе фото'),
        ]
    )
    await message.answer_media_group(media=album)

Пользователь получает два фото в одном сообщении (альбом).

Пример 4: Использование Redis для хранения состояний (FSM Storage)

from aiogram.fsm.storage.redis import RedisStorage
import redis

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
storage = RedisStorage(redis_client)
dp = Dispatcher(storage=storage)

# Теперь FSM состояния сохраняются в Redis, что полезно при масштабировании.

RedisStorage позволяет хранить данные между перезапусками бота и в горизонтально масштабируемой среде.

Пример 5: Фильтры на основе пользовательских прав (админка)

from aiogram.filters import Filter
from aiogram.types import Message

ADMIN_IDS = [123456789]  # ID администраторов

class AdminFilter(Filter):
    async def __call__(self, message: Message) -> bool:
        return message.from_user.id in ADMIN_IDS

@router.message(Command('admin'), AdminFilter())
async def admin_panel(message: Message):
    await message.answer('Добро пожаловать в панель администратора.')

Команда /admin срабатывает только у пользователей с ID из списка ADMIN_IDS. Для остальных хендлер игнорируется.

Пример 6: Обработка ошибок через ErrorEvent

from aiogram.types import ErrorEvent

@dp.error()
async def error_handler(event: ErrorEvent):
    print(f'Ошибка: {event.exception}')
    await event.update.message.answer('Произошла внутренняя ошибка.')

Глобальный обработчик ошибок перехватывает все необработанные исключения в хендлерах. Бот не падает, а уведомляет пользователя.

Пример 7: Чат-бот с использованием GPT через Aiogram

import openai
from aiogram import types

@router.message(Command('ask'))
async def ask_gpt(message: types.Message):
    user_text = message.text.replace('/ask ', '')
    response = openai.ChatCompletion.create(
        model='gpt-3.5-turbo',
        messages=[{'role': 'user', 'content': user_text}]
    )
    reply = response.choices[0].message.content
    await message.answer(reply)

Интеграция с внешними API - частое применение ботов. Важно установить отдельно библиотеку openai и настроить ключ.