Конечные автоматы, кастомные storage, Redis, контексты диалогов
Проблема: Пользователь начал регистрацию: ввёл имя, но бот не знает, что делать с следующим сообщением. Это email? Возраст? Комментарий? Без контекста каждое сообщение обрабатывается изолированно.
Типичные ошибки:
- Хранение состояния в переменной — теряется между сообщениями
- Проверка "последней команды" в каждом обработчике — дублирование кода
- Нет способа отменить многошаговый процесс
Решение: FSM (Finite State Machine) хранит состояние диалога между сообщениями. Бот "помнит", на каком этапе находится пользователь, и обрабатывает сообщения соответственно.
Зачем это нужно: Любая многошаговая форма (регистрация, заказ, опрос) требует FSM. Без неё код превращается в спагетти из проверок.
FSM (Finite State Machine, конечный автомат) — модель, позволяющая боту "помнить" текущий этап диалога с пользователем.
# ❌ Так не делайте: хранение состояния в переменной
user_state = {} # Потеряется при перезапуске бота!
@router.message(Command('register'))
async def cmd_register(msg: Message):
user_state[msg.from_user.id] = 'name'
await msg.answer('Введите имя:')
@router.message()
async def any_message(msg: Message):
# ⚠️ Проблема: user_state может не существовать
# ⚠️ Проблема: нет структуры для данных
if user_state.get(msg.from_user.id) == 'name':
name = msg.text
# ...Проблемы такого подхода:
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.context import FSMContext
class Registration(StatesGroup):
"""Сценарий регистрации пользователя."""
name = State()
email = State()
age = State()
@router.message(Command('register'))
async def cmd_register(msg: Message, state: FSMContext):
"""
Начало регистрации.
💡 Зачем state.set_state(): Бот "запоминает", что пользователь
начал регистрацию и ждёт ввода имени.
"""
await state.set_state(Registration.name)
await msg.answer('Введите имя:')
@router.message(Registration.name)
async def process_name(msg: Message, state: FSMContext):
"""
Обработка имени.
💡 Зачем: Фильтр Registration.name гарантирует, что
обработчик сработает только для пользователей в состоянии name.
"""
# Сохраняем данные в состоянии
await state.update_data(name=msg.text)
# Переходим к следующему шагу
await state.set_state(Registration.email)
await msg.answer('Введите email:')💡 Зачем StatesGroup: Группировка состояний логически связывает этапы одного сценария.
StatesGroup — группа состояний для одного сценария. State — отдельное состояние внутри группы.
from aiogram.fsm.state import State, StatesGroup
class Registration(StatesGroup):
"""
Сценарий регистрации пользователя.
💡 Зачем:
- name → email → age → confirm — линейный сценарий
- Можно вернуться на шаг назад
- Можно отменить в любой момент
"""
name = State()
email = State()
age = State()
confirm = State()
class OrderForm(StatesGroup):
"""
Сценарий оформления заказа.
💡 Зачем: Отдельная группа для другого сценария.
Состояния не пересекаются с Registration.
"""
product = State()
quantity = State()
address = State()
payment = State()
class FeedbackForm(StatesGroup):
"""Сценарий обратной связи."""
rating = State()
comment = State()
contact = State()⚠️ Антипаттерн: Не используйте одно состояние для разных сценариев.
nameв Registration иnameв ProfileForm — это разные состояния.
class Profile(StatesGroup):
"""
Вложенные группы для сложных сценариев.
💡 Зачем: Логическая группировка внутри сценария.
"""
class Personal(StatesGroup):
name = State()
birthday = State()
class Contact(StatesGroup):
email = State()
phone = State()
# Использование
@router.message(Profile.Personal.name)
async def process_name(msg: Message, state: FSMContext):
...💡 Совет: Избегайте вложенности глубже 2 уровней. Это усложняет отладку.
FSMContext — объект для управления состоянием пользователя.
from aiogram.fsm.context import FSMContext
@router.message(Command('start'))
async def cmd_start(msg: Message, state: FSMContext):
"""
Управление состоянием.
💡 Методы:
- set_state(): установить состояние
- get_state(): получить текущее
- update_data(): сохранить данные
- get_data(): получить все данные
- clear(): очистить состояние и данные
"""
# Установить состояние
await state.set_state(Registration.name)
# Получить текущее состояние
current_state = await state.get_state()
# 'Registration:name'
# Очистить состояние (завершить сценарий)
await state.set_state(None)
# Очистить все данные
await state.clear()
@router.message(Registration.name)
async def process_name(msg: Message, state: FSMContext):
# Сохранить данные
await state.update_data(name=msg.text)
# Получить все данные
data = await state.get_data()
# {'name': 'Иван'}
# Получить конкретное поле
name = data.get('name')
# Изменить данные
await state.update_data(name='Алексей')
# Удалить поле (установить None)
await state.update_data(name=None)⚠️ Важно:
update_data()не перезаписывает все данные, а обновляет указанные поля.
from aiogram import Router, F
from aiogram.types import Message
from aiogram.filters import Command, StateFilter
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from bot.keyboards.inline import build_confirm_keyboard
router = Router()
class Registration(StatesGroup):
name = State()
email = State()
age = State()
confirm = State()
@router.message(Command('register'))
async def cmd_register(msg: Message, state: FSMContext):
"""Начало регистрации."""
await state.set_state(Registration.name)
await msg.answer('📝 Введите ваше имя:')
@router.message(Registration.name, F.text.func(lambda x: len(x) >= 2))
async def process_name(msg: Message, state: FSMContext):
"""
Обработка имени.
⚠️ Важно: Фильтр длины текста отсеивает невалидный ввод.
Обработчик не сработает для имени из 1 символа.
"""
await state.update_data(name=msg.text)
await state.set_state(Registration.email)
await msg.answer('📧 Введите email:')
@router.message(Registration.name) # Слишком короткое имя
async def invalid_name(msg: Message):
"""
Обработчик для невалидного имени.
💡 Зачем: Второй обработчик срабатывает, если первый не прошёл фильтр.
"""
await msg.answer('Имя должно быть не менее 2 символов. Попробуйте снова:')
@router.message(Registration.email, F.text.contains('@'))
async def process_email(msg: Message, state: FSMContext):
"""Обработка email."""
await state.update_data(email=msg.text)
await state.set_state(Registration.age)
await msg.answer('🎂 Введите возраст:')
@router.message(Registration.email) # Неверный email
async def invalid_email(msg: Message):
await msg.answer('Email должен содержать @. Попробуйте снова:')
@router.message(Registration.age, F.text.func(lambda x: x.isdigit()))
async def process_age(msg: Message, state: FSMContext):
"""Обработка возраста."""
age = int(msg.text)
if age < 1 or age > 120:
await msg.answer('Возраст должен быть от 1 до 120. Попробуйте снова:')
return
await state.update_data(age=age)
await state.set_state(Registration.confirm)
# Показываем подтверждение
data = await state.get_data()
text = f"""
Проверьте данные:
Имя: {data['name']}
Email: {data['email']}
Возраст: {data['age']}
"""
await msg.answer(text, reply_markup=build_confirm_keyboard())
@router.message(Registration.confirm, F.text == 'Подтвердить')
async def confirm_registration(msg: Message, state: FSMContext, user_repo):
"""
Подтверждение регистрации.
💡 Зачем state.clear(): Завершаем сценарий, очищаем память.
"""
data = await state.get_data()
# Создаём пользователя в БД
await user_repo.create(
telegram_id=msg.from_user.id,
username=msg.from_user.username,
name=data['name'],
email=data['email'],
age=data['age']
)
await state.clear()
await msg.answer('✅ Регистрация завершена!')
@router.message(Registration.confirm, F.text == 'Изменить')
async def edit_registration(msg: Message, state: FSMContext):
"""Возврат к редактированию."""
await state.set_state(Registration.name)
await msg.answer('✏️ Введите новое имя:')
@router.message(Registration.confirm, F.text == 'Отмена')
async def cancel_registration(msg: Message, state: FSMContext):
"""Отмена регистрации."""
await state.clear()
await msg.answer('❌ Регистрация отменена')⚠️ Антипаттерн: Не забывайте
state.clear()после завершения сценария. Иначе пользователь останется в состоянии и следующие сообщения будут обрабатываться неправильно.
from aiogram.fsm.storage.memory import MemoryStorage
storage = MemoryStorage()
dp = Dispatcher(storage=storage)| Плюсы | Минусы |
|---|---|
| Не требует внешних зависимостей | Данные теряются при перезапуске |
| Быстрое | Не работает при нескольких инстансах бота |
| Удобно для локальной разработки | Не для продакшена |
💡 Когда использовать: Локальная разработка, тесты, прототипы.
from redis.asyncio import Redis
from aiogram.fsm.storage.redis import RedisStorage
redis = Redis.from_url('redis://localhost:6379/0')
storage = RedisStorage(redis=redis, key_prefix='fsm')
dp = Dispatcher(storage=storage)| Плюсы | Минусы |
|---|---|
| Персистентное хранение | Требует Redis |
| Работает с несколькими инстансами | Дополнительная инфраструктура |
| Быстрый доступ |
💡 Зачем key_prefix: Изоляция ключей в Redis для разных ботов/окружений.
from aiogram.fsm.storage.base import BaseStorage, StateType, StorageKey
from typing import Dict, Any, Optional
from sqlalchemy.ext.asyncio import AsyncSession
class DatabaseStorage(BaseStorage):
"""
Хранение состояний в БД.
💡 Зачем:
- Не нужен Redis
- Данные в той же БД, что и пользователи
- Бэкапы автоматически
⚠️ Антипаттерн: Не используйте для высоконагруженных ботов.
Redis быстрее для частых операций записи/чтения.
"""
def __init__(self, session_factory):
self.session_factory = session_factory
async def set_state(self, key: StorageKey, state: StateType = None) -> None:
async with self.session_factory() as session:
# Сохраняем состояние в БД
...
async def get_state(self, key: StorageKey) -> Optional[str]:
async with self.session_factory() as session:
# Получаем состояние из БД
...
async def set_data(self, key: StorageKey, data: Dict[str, Any]) -> None:
async with self.session_factory() as session:
# Сохраняем данные в БД
...
async def get_data(self, key: StorageKey) -> Dict[str, Any]:
async with self.session_factory() as session:
# Получаем данные из БД
...Проблема: В зависимости от ответа пользователя нужно идти по разным веткам сценария.
Решение: Условная логика в обработчиках для перехода к разным состояниям.
from aiogram.fsm.state import State, StatesGroup
from aiogram.types import CallbackQuery
class Survey(StatesGroup):
"""
Опрос с ветвлением.
💡 Зачем: Разные вопросы для разных возрастных групп.
"""
age_group = State()
experience = State()
skills = State()
goals = State()
@router.callback_query(F.data == 'start_survey')
async def start_survey(cb: CallbackQuery, state: FSMContext):
await state.set_state(Survey.age_group)
await cb.message.edit_text(
'Выберите возраст:',
reply_markup=build_age_keyboard()
)
@router.callback_query(Survey.age_group, F.data.startswith('age_'))
async def process_age(cb: CallbackQuery, state: FSMContext):
"""
Обработка возраста с ветвлением.
💡 Зачем: Молодым пользователям не задаём вопрос об опыте.
"""
age_group = cb.data.split('_')[1]
await state.update_data(age_group=age_group)
# Ветвление: разные вопросы для разных возрастов
if age_group in ['under_18', '18_25']:
# Молодые → сразу к навыкам
await state.set_state(Survey.skills)
await cb.message.edit_text(
'Какие навыки вы хотите развить?',
reply_markup=build_skills_keyboard()
)
else:
# Старшие → вопрос об опыте
await state.set_state(Survey.experience)
await cb.message.edit_text(
'Ваш опыт работы?',
reply_markup=build_experience_keyboard()
)
@router.callback_query(Survey.experience, F.data.startswith('exp_'))
async def process_experience(cb: CallbackQuery, state: FSMContext):
"""Продолжение после опыта."""
await state.update_data(experience=cb.data.split('_')[1])
await state.set_state(Survey.skills)
await cb.message.edit_text(
'Какие навыки вы хотите развить?',
reply_markup=build_skills_keyboard()
)💡 Совет: Визуализируйте сценарий на бумаге перед реализацией. Это поможет выявить сложные ветвления.
Проблема: Пользователь начал регистрацию и исчез. Состояние остаётся в памяти бесконечно.
Решение: Таймаут сбрасывает состояние после периода неактивности.
import asyncio
from aiogram.fsm.context import FSMContext
async def state_timeout(state: FSMContext, delay: int):
"""
Сбрасывает состояние через delay секунд.
⚠️ Антипаттерн: asyncio.create_task() без обработки ошибок.
Задача может упасть незаметно.
"""
await asyncio.sleep(delay)
current_state = await state.get_state()
if current_state:
await state.clear()
# Уведомить пользователя
...
@router.message(Command('order'))
async def cmd_order(msg: Message, state: FSMContext):
await state.set_state(OrderForm.product)
await msg.answer('Выберите товар (у вас 5 минут):')
# Запускаем таймаут
# ⚠️ Важно: В продакшене используйте Celery (см. "Масштабирование")
asyncio.create_task(state_timeout(state, 300))⚠️ Антипаттерн: Не полагайтесь на asyncio.create_task() в продакшене. Используйте Celery beat или Redis key expiry для надёжности.
class Payment(StatesGroup):
"""
Сценарий оплаты.
💡 Зачем:
- method → card_number → confirm — линейный процесс
- Разные пути для карты и криптовалюты
- Возможность отмены на любом этапе
"""
method = State()
card_number = State()
confirm = State()
@router.callback_query(F.data == 'pay')
async def start_payment(cb: CallbackQuery, state: FSMContext):
await state.set_state(Payment.method)
await cb.message.edit_text(
'Выберите способ оплаты:',
reply_markup=build_payment_keyboard()
)
@router.callback_query(Payment.method, F.data == 'card')
async def choose_card(cb: CallbackQuery, state: FSMContext):
await state.set_state(Payment.card_number)
await cb.message.edit_text(
'Введите номер карты (или нажмите /cancel):'
)
@router.callback_query(Payment.method, F.data == 'crypto')
async def choose_crypto(cb: CallbackQuery, state: FSMContext):
"""
Альтернативный путь для криптовалюты.
💡 Зачем: Пропускаем ввод карты, сразу к подтверждению.
"""
await state.update_data(method='crypto')
await state.set_state(Payment.confirm)
await cb.message.edit_text(
'Отправляем на криптокошелёк...\n'
'Адрес: 0x1234567890abcdef\n\n'
'Подтвердить оплату?',
reply_markup=build_confirm_keyboard()
)
@router.message(Payment.card_number, F.text.func(lambda x: len(x) == 16 and x.isdigit()))
async def process_card(msg: Message, state: FSMContext):
"""Валидация номера карты."""
await state.update_data(card_number=msg.text)
await state.set_state(Payment.confirm)
await msg.answer(
f'Оплата картой {msg.text[-4:]}.\nПодтвердить?',
reply_markup=build_confirm_keyboard()
)
@router.message(Payment.card_number)
async def invalid_card(msg: Message):
"""Неверный номер карты."""
await msg.answer('Неверный номер карты. Попробуйте снова:')
@router.callback_query(Payment.confirm, F.data == 'confirm_payment')
async def confirm_payment(cb: CallbackQuery, state: FSMContext, payment_service):
"""
Подтверждение оплаты.
⚠️ Важно: state.clear() только после успешной оплаты.
При ошибке сохраняем состояние для повторной попытки.
"""
data = await state.get_data()
try:
await payment_service.process_payment(
user_id=cb.from_user.id,
method=data.get('method', 'card'),
card_number=data.get('card_number')
)
await state.clear()
await cb.message.edit_text('✅ Оплата прошла успешно!')
except PaymentError as e:
await cb.message.edit_text(f'❌ Ошибка оплаты: {e}')
# Не очищаем состояние — пользователь может попробовать сноваПроблема: Пользователь передумал заполнять форму. Как выйти из сценария?
Решение: Универсальный обработчик команды /cancel для любого состояния.
from aiogram.filters import Command
@router.message(Command('cancel'), ~StateFilter(None))
async def cancel_state(msg: Message, state: FSMContext):
"""
Отмена текущего состояния.
💡 Зачем ~StateFilter(None): Срабатывает только если есть активное состояние.
"""
current_state = await state.get_state()
if current_state:
await state.clear()
await msg.answer('❌ Действие отменено')
@router.callback_query(F.data == 'cancel', ~StateFilter(None))
async def cancel_callback(cb: CallbackQuery, state: FSMContext):
"""Отмена через callback."""
await state.clear()
await cb.message.edit_text('❌ Действие отменено')💡 Совет: Добавьте кнопку "Отмена" в клавиатуру каждого шага многошаговой формы.
from aiogram import BaseMiddleware
from aiogram.types import Update
from aiogram.fsm.context import FSMContext
class StateTimeoutMiddleware(BaseMiddleware):
"""
Сбрасывает состояние после 30 минут неактивности.
💡 Зачем:
- Очистка "зависших" состояний
- Освобождение памяти
⚠️ Антипаттерн: Не используйте для критичных сценариев (оплата).
Там нужен явный контроль таймаута.
"""
async def __call__(
self,
handler,
event: Update,
data: dict
):
state: FSMContext = data.get('state')
if state:
last_activity = await state.get_data().get('last_activity')
if last_activity:
import time
if time.time() - last_activity > 1800: # 30 минут
await state.clear()
await event.message.answer(
'Сессия истекла. Начните сначала.'
)
return
await state.update_data(last_activity=time.time())
return await handler(event, data)| Тема | Как связана |
|---|---|
| Обработчики | StateFilter для фильтрации по состоянию |
| Архитектура | Storage настраивается в dispatcher |
| Базы данных | Кастомное хранилище в БД |
| Масштабирование | Redis для shared state между инстансами |
| Inline-клавиатуры | Навигация в многошаговых формах |
→ Базы данных — сохранение данных формы в БД
→ Масштабирование — Celery для таймаутов и отложенных задач
→ Тестирование — тесты FSM-сценариев
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.