Кастомные фильтры, приоритеты обработчиков, флаги, цепочки ответственности
Проблема: Вы хотите, чтобы команда
/adminработала только для админов, сообщения с фото обрабатывались отдельно от текста, а команда/sum 1 2 3автоматически суммировала числа. Без фильтров придётся писать проверки в каждом обработчике — код раздувается, логика дублируется, тестировать невозможно.Решение: Фильтры выносят условия обработки в декларативную форму. Обработчик получает только те сообщения, которые ему нужны. Это делает код чище, упрощает тестирование и позволяет переиспользовать логику.
Обработчик (handler) — асинхронная функция, которая вызывается при совпадении обновления с определёнными условиями.
💡 Зачем это нужно: Обработчики — это точки входа для вашей логики. Каждый сценарий (команда, кнопка, файл) имеет свой обработчик. Правильное разделение обработчиков упрощает поддержку: изменили одну команду — не сломали другие.
from aiogram import Router, F
from aiogram.types import Message
from aiogram.filters import Command
router = Router()
@router.message(Command('start'))
async def cmd_start(msg: Message):
"""
Обработчик команды /start.
Зачем: Это первая точка контакта с пользователем.
Здесь можно проверить, зарегистрирован ли пользователь,
и показать приветственное сообщение.
"""
await msg.answer('Привет! Я бот.')⚠️ Важно: Имя функции (
cmd_start) не влияет на логику, но помогает при отладке и в логах. Используйте префиксы:cmd_для команд,cb_для callback,msg_для сообщений.
Фильтр определяет, должен ли обработчик реагировать на обновление.
🔗 Связь с другими темами: Кастомные фильтры выносятся в модуль
bot/filters/и используются вместе с Middleware для сквозной логики (например, проверка прав доступа).
| Сценарий | Фильтр | Пример использования |
|---|---|---|
| Пользователь ввёл команду | Command | /start, /help, /admin |
| Нажал кнопку в меню | Text | "Меню", "Профиль", "Настройки" |
| Прислал фото/документ | F.photo, F.document | Загрузка аватарки, импорта данных |
| Нужно проверить состояние | StateFilter | Диалог регистрации, оформления заказа |
| Только для определённого чата | ChatTypeFilter | Админ-команды только в личных сообщениях |
| Сложное условие | Magic Filter F | Комбинация нескольких проверок |
Проблема: Пользователь вводит /sum 1 2 3 4, и вы хотите получить сумму чисел. Как извлечь аргументы команды?
Решение: Используйте CommandObject для доступа к аргументам и magic для валидации.
from aiogram.filters import Command, CommandObject
@router.message(Command('echo', magic=F.args.func(lambda x: len(x) > 0)))
async def cmd_echo(msg: Message, command: CommandObject):
"""
Эхо-команда с аргументами.
Зачем: Проверка наличия аргументов до вызова обработчика.
Если аргументов нет — обработчик не сработает.
"""
# command.args — аргументы после команды
await msg.answer(f'Вы ввели: {command.args}')
@router.message(Command('sum', magic=F.args.func(lambda x: x.strip())))
async def cmd_sum(msg: Message, command: CommandObject):
"""
Суммирует числа: /sum 1 2 3 4
Зачем: Автоматическая валидация входных данных.
Фильтр отсечёт команды без аргументов.
"""
try:
numbers = list(map(int, command.args.split()))
result = sum(numbers)
await msg.answer(f'Сумма: {result}')
except ValueError:
await msg.answer('Пожалуйста, введите числа')⚠️ Антипаттерн: Не делайте сложную валидацию внутри обработчика. Если аргументы обязательны — используйте
magicдля проверки до вызова функции.
Проблема: Пользователь нажал кнопку "Меню" в reply-клавиатуре. Как отличить это от обычного сообщения с текстом "Меню"?
Решение: Фильтр Text точно совпадает с текстом кнопки.
from aiogram.filters import Text
@router.message(Text(text='Меню', ignore_case=True))
async def menu_handler(msg: Message):
"""
Обработчик нажатия кнопки 'Меню'.
Зачем: ignore_case=True позволяет ловить 'меню', 'МЕНЮ', 'Меню'.
Полезно, когда пользователь вводит текст вручную.
"""
await msg.answer('Открываю меню...')
@router.message(Text(startswith='Настройка'))
async def settings_handler(msg: Message):
"""
Обработчик для раздела настроек.
Зачем: startswith позволяет ловить 'Настройка профиля',
'Настройка уведомлений' и т.д. одним фильтром.
"""
await msg.answer('Раздел настроек')
@router.message(Text(endswith='помощь'))
async def help_handler(msg: Message):
"""
Обработчик помощи.
Зачем: endswith ловит 'Мне нужна помощь', 'Вызвать помощь'.
"""
await msg.answer('Чем помочь?')💡 Совет: Для inline-кнопок используйте callback_data вместо Text — это надёжнее (см. тему "Inline-клавиатуры").
Проблема: Нужно проверить, что сообщение от админа, содержит текст больше 10 символов и отправлено в будний день. Писать три проверки в обработчике?
Решение: Magic Filter позволяет комбинировать условия декларативно.
from aiogram import F
from aiogram.types import Message
# Проверка атрибута
@router.message(F.text == 'Привет')
async def hello_handler(msg: Message):
"""
Зачем: Точное совпадение текста.
Когда: Для команд меню, ответов на вопросы.
"""
await msg.answer('Привет!')
# Проверка вложенного атрибута
@router.message(F.from_user.id == 123456789)
async def admin_handler(msg: Message):
"""
Зачем: Проверка конкретного пользователя.
Когда: Для админ-команд, whitelist-доступа.
⚠️ Антипаттерн: Не хардкодьте ID в фильтрах.
Используйте кастомный фильтр is_admin (см. ниже).
"""
await msg.answer('Привет, админ!')
# Проверка наличия атрибута
@router.message(F.photo)
async def photo_handler(msg: Message):
"""
Зачем: Обработка только фото.
Когда: Загрузка аватарок, модерация контента.
"""
await msg.answer('Вижу фото!')
# Отрицание
@router.message(~F.text.startswith('/'))
async def not_command_handler(msg: Message):
"""
Зачем: Игнорировать команды.
Когда: Обработка обычных сообщений в чате.
"""
await msg.answer('Это не команда')from aiogram import F
# Логические операторы
@router.message(F.text == 'Да' | F.text == 'Нет')
async def yes_no_handler(msg: Message):
"""
Зачем: Один обработчик для нескольких вариантов.
Когда: Подтверждение/отмена, да/нет.
"""
await msg.answer('Выбор сделан')
# Комбинирование
@router.message(F.text.len() > 5 & F.text.len() < 100)
async def medium_text_handler(msg: Message):
"""
Зачем: Ограничение длины текста.
Когда: Валидация отзывов, комментариев.
⚠️ Антипаттерн: Не используйте сложные условия
в декораторе — вынесите в кастомный фильтр.
"""
await msg.answer('Текст средней длины')
# Методы строк
@router.message(F.text.startswith('/'))
@router.message(F.text.endswith('?'))
@router.message(F.text.contains('спасибо'))
async def text_handlers(msg: Message):
"""
Зачем: Быстрая проверка содержимого.
"""
await msg.answer('Получено')
# Методы списков
@router.message(F.animation.func(lambda x: x.file_size < 1024 * 1024))
async def small_gif_handler(msg: Message):
"""
Зачем: Ограничение размера файла.
Когда: Экономия трафика, места на диске.
"""
await msg.answer('Лёгкий GIF')
# Доступ по индексу
@router.message(F.entities[0].type == 'bot_command')
async def command_entity_handler(msg: Message):
"""
Зачем: Проверка типа entity.
Когда: Обработка команд в тексте.
⚠️ Антипаттерн: Доступ по индексу [0] упадёт,
если entities пуст. Используйте F.entities.first()
или проверку len(F.entities) > 0.
"""
await msg.answer('Обнаружена команда')from aiogram import F
from datetime import datetime
def is_weekend(dt: datetime) -> bool:
"""Проверяет, выходной ли день."""
return dt.weekday() >= 5
@router.message(F.date.func(is_weekend))
async def weekend_handler(msg: Message):
"""
Зачем: Разная логика для будней/выходных.
Когда: Рабочий бот, уведомления только в рабочее время.
"""
await msg.answer('Сейчас выходной!')
# Лямбда-функции
@router.message(F.text.func(lambda x: x.isdigit()))
async def digit_handler(msg: Message):
"""
Зачем: Проверка формата.
Когда: Ввод возраста, количества, кода.
"""
await msg.answer(f'Число: {msg.text}')💡 Совет: Если функция используется в нескольких местах — вынесите в кастомный фильтр-класс.
Проблема: Вы проверяете user_id in admin_ids в пяти обработчиках. Изменили список админов — правите пять файлов.
Решение: Кастомный фильтр инкапсулирует логику проверки. Изменили в одном месте — работает везде.
📁 Структура проекта: Сохраняйте фильтры в
bot/filters/для переиспользования.
# bot/filters/is_admin.py
from aiogram.types import Message, CallbackQuery
from bot.config import settings
async def admin_filter(msg: Message | CallbackQuery) -> bool:
"""
Проверяет, является ли пользователь админом.
Зачем: Централизованная проверка прав.
Когда: Админ-команды, доступ к статистике, модерация.
"""
user_id = (msg.from_user.id if hasattr(msg, 'from_user')
else msg.from_user.id)
return user_id in settings.admin_ids
# Использование
@router.message(admin_filter)
async def admin_handler(msg: Message):
await msg.answer('Привет, админ!')⚠️ Антипаттерн: Не вызывайте
admin_filter()вручную в обработчике. Используйте как декоратор — aiogram сам вызовет фильтр.
# bot/filters/is_subscribed.py
from aiogram import BaseFilter
from aiogram.types import Message, CallbackQuery
from bot.services.api_client import APIClient
class IsSubscribedFilter(BaseFilter):
"""
Проверяет подписку пользователя на канал.
Зачем: Ограничение доступа для неподписанных.
Когда: Контент только для подписчиков, закрытые клубы.
🔗 Связь с другими темами: Для массовой проверки используйте
кэширование (Redis) для снижения нагрузки на API Telegram.
"""
def __init__(self, channel_id: int):
self.channel_id = channel_id
async def __call__(self, event: Message | CallbackQuery) -> bool:
user_id = event.from_user.id
# Проверка подписки через get_chat_member
return await self.check_subscription(user_id)
async def check_subscription(self, user_id: int) -> bool:
# Логика проверки (через бота или API)
...
# Использование
@router.message(IsSubscribedFilter(channel_id=-1001234567890))
async def subscribed_handler(msg: Message):
await msg.answer('Спасибо за подписку!')🔗 Связь с другими темами: Этот фильтр использует Dependency Injection для доступа к БД. См. тему "Архитектурные паттерны".
# bot/filters/is_premium_user.py
from aiogram import BaseFilter
from aiogram.types import Message
from sqlalchemy.ext.asyncio import AsyncSession
from bot.repositories.user import UserRepository
class IsPremiumUser(BaseFilter):
"""
Проверяет, есть ли у пользователя премиум-подписка.
Зачем: Разделение функционала для платных/бесплатных пользователей.
Когда: Premium-контент, приоритетная поддержка, расширенные функции.
"""
def __init__(self, required: bool = True):
self.required = required
async def __call__(
self,
msg: Message,
session: AsyncSession,
user_repo: UserRepository
) -> bool:
user = await user_repo.get_by_telegram_id(msg.from_user.id)
is_premium = user and user.is_premium
return is_premium if self.required else not is_premium
# Использование
@router.message(IsPremiumUser())
async def premium_handler(msg: Message):
await msg.answer('Премиум-контент')
@router.message(IsPremiumUser(required=False))
async def non_premium_handler(msg: Message):
await msg.answer('Доступно только для премиум')⚠️ Критично важно: Это одна из самых частых ошибок новичков.
Проблема: Вы зарегистрировали общий обработчик всех сообщений, а потом — обработчик команды /start. Команда не работает.
Почему: aiogram обрабатывает обработчики в порядке регистрации. Первый совпавший обработчик выполняется, остальные игнорируются.
# ❌ НЕПРАВИЛЬНО: общий обработчик перехватит все сообщения
@router.message()
async def catch_all(msg: Message):
await msg.answer('Получено')
@router.message(Command('start'))
async def cmd_start(msg: Message):
await msg.answer('Start') # Никогда не вызовется!
# ✅ ПРАВИЛЬНО: специфичные обработчики раньше общих
@router.message(Command('start'))
async def cmd_start(msg: Message):
await msg.answer('Start')
@router.message(Command('help'))
async def cmd_help(msg: Message):
await msg.answer('Help')
@router.message() # Общий обработчик в конце
async def catch_all(msg: Message):
await msg.answer('Получено')💡 Правило: От специфичного к общему. Сначала команды, потом текст, потом catch-all.
Проблема: У вас 50 обработчиков в одном файле. Порядок регистрации легко нарушить.
Решение: Группируйте обработчики по роутерам и регистрируйте роутеры в нужном порядке.
# bot/dispatcher.py
from bot.handlers import high_priority, medium_priority, low_priority
dp = Dispatcher()
# Регистрируем роутеры в порядке приоритета
dp.include_router(high_priority.router) # Команды, callback
dp.include_router(medium_priority.router) # Текстовые сообщения
dp.include_router(low_priority.router) # Catch-all🔗 Связь с другими темами: Роутеры — часть модульной архитектуры. См. тему "Архитектурные паттерны".
Проблема: Нужно передать метаданные в обработчик: требует ли он авторизации, какой rate limit применять. Писать отдельные фильтры для каждого случая?
Решение: Флаги — механизм передачи метаданных через декоратор.
from aiogram import Router, F
from aiogram.types import Message
from aiogram.dispatcher.flags import get_flag
router = Router()
@router.message(F.text == 'admin', flags={'require_admin': True})
async def admin_only(msg: Message):
"""
Зачем: Флаг указывает, что обработчик требует прав админа.
⚠️ Антипаттерн: Не полагайтесь только на флаги.
Флаг — это метаданные, а не защита.
Проверку прав делайте в middleware или обработчике.
"""
# Извлекаем флаг внутри обработчика
require_admin = get_flag(msg, 'require_admin', default=False)
if require_admin:
# Проверка прав
if not is_admin(msg.from_user.id):
return
await msg.answer('Админ-панель')🔗 Связь с другими темами: Middleware — сквозной механизм. См. тему "Архитектурные паттерны".
# bot/middleware/flags.py
from typing import Callable, Dict, Any
from aiogram import BaseMiddleware
from aiogram.types import Update
from aiogram.dispatcher.flags import get_flag
class FlagMiddleware(BaseMiddleware):
"""
Обрабатывает флаги перед вызовом обработчика.
Зачем: Централизованная обработка метаданных.
Когда: Аутентификация, rate limiting, логирование.
"""
async def __call__(
self,
handler: Callable,
event: Update,
data: Dict[str, Any]
) -> Any:
# Извлекаем флаги из контекста
require_auth = get_flag(event, 'require_auth', default=False)
rate_limit = get_flag(event, 'rate_limit', default=10)
if require_auth:
user = data.get('user')
if not user:
return # Прерываем, если нет авторизации
# Передаем флаги в data
data['flags'] = {
'require_auth': require_auth,
'rate_limit': rate_limit
}
return await handler(event, data)
# Использование
@router.message(
F.text == 'profile',
flags={'require_auth': True, 'rate_limit': 5}
)
async def profile_handler(msg: Message):
flags = msg.flags # Или из контекста
await msg.answer('Профиль')Проблема: Нужно, чтобы обработчик срабатывал только для команды /register, от пользователя с языком 'ru' и не в состоянии формы.
Решение: Несколько фильтров в декораторе комбинируются через логическое И.
from aiogram import F
from aiogram.filters import Command, StateFilter
from aiogram.fsm.state import State, StatesGroup
class Form(StatesGroup):
name = State()
age = State()
# Все фильтры должны совпасть
@router.message(
Command('register'),
F.from_user.language_code == 'ru',
~StateFilter(Form.name) # Не в состоянии Form.name
)
async def cmd_register(msg: Message):
"""
Зачем: Комбинация условий для точного контроля.
Когда:
- Command: только для команды
- language_code: локализация
- ~StateFilter: игнорировать, если пользователь в форме
"""
await msg.answer('Начинаем регистрацию')⚠️ Антипаттерн: Не перегружайте декоратор условиями. Если фильтров больше 3-4 — вынесите в кастомный фильтр-класс.
from aiogram import F
from aiogram.types import ContentType
@router.message(F.content_type == ContentType.PHOTO)
async def photo_handler(msg: Message):
"""
Зачем: Обработка только фото.
Когда: Загрузка аватарок, модерация изображений.
"""
photo = msg.photo[-1] # Лучшее качество
await msg.answer(f'Фото получено, file_id: {photo.file_id}')
@router.message(F.content_type == ContentType.DOCUMENT)
async def doc_handler(msg: Message):
"""
Зачем: Обработка документов.
Когда: Импорт данных, загрузка файлов.
"""
await msg.answer(f'Документ: {msg.document.file_name}')
@router.message(F.content_type == ContentType.VOICE)
async def voice_handler(msg: Message):
"""
Зачем: Обработка голосовых.
Когда: Транскрибация, голосовые команды.
"""
await msg.answer('Голосовое сообщение получено')
# Любой медиа-контент
@router.message(F.photo | F.document | F.video | F.audio)
async def media_handler(msg: Message):
"""
Зачем: Обработка любого медиа.
Когда: Универсальный загрузчик файлов.
"""
await msg.answer('Медиа получено')from aiogram import F
from datetime import datetime, time
async def is_business_hours(msg: Message) -> bool:
"""
Проверяет, рабочее ли время (9:00 - 18:00).
Зачем: Ограничение доступа по времени.
Когда: Рабочий бот, уведомления только в рабочее время.
⚠️ Антипаттерн: Не используйте datetime.now()
без учёта часового пояса пользователя.
"""
now = datetime.now().time()
return time(9, 0) <= now <= time(18, 0)
async def is_weekday(msg: Message) -> bool:
"""Проверяет, будний ли день."""
return datetime.now().weekday() < 5
@router.message(is_business_hours & is_weekday)
async def business_hours_handler(msg: Message):
await msg.answer('Мы работаем!')
@router.message(~is_business_hours | ~is_weekday)
async def off_hours_handler(msg: Message):
await msg.answer('Мы сейчас не работаем. Оставьте сообщение.')💡 Совет: Для работы с часовыми поясами используйте
pytzилиzoneinfo.
Проблема: Нужно проверить длину текста, язык пользователя, наличие премиума и время отправки. Четыре отдельных фильтра загромождают код.
Решение: Один фильтр-класс с параметрами.
# bot/filters/complex_filter.py
from aiogram import BaseFilter, F
from aiogram.types import Message
from datetime import datetime
class ComplexFilter(BaseFilter):
"""
Сложный фильтр с несколькими условиями.
Зачем: Инкапсуляция сложной логики проверки.
Когда: Многофакторная валидация сообщений.
"""
def __init__(
self,
min_text_length: int = 1,
allowed_languages: list[str] | None = None,
require_premium: bool = False
):
self.min_text_length = min_text_length
self.allowed_languages = allowed_languages or ['ru', 'en']
self.require_premium = require_premium
async def __call__(self, msg: Message) -> bool:
# Проверка длины текста
if not msg.text or len(msg.text) < self.min_text_length:
return False
# Проверка языка
if msg.from_user.language_code not in self.allowed_languages:
return False
# Проверка премиума
if self.require_premium and not msg.from_user.is_premium:
return False
# Проверка времени (не ночью)
hour = datetime.now().hour
if hour < 6 or hour > 23:
return False
return True
# Использование
@router.message(ComplexFilter(
min_text_length=5,
allowed_languages=['ru'],
require_premium=False
))
async def filtered_handler(msg: Message):
await msg.answer('Сообщение прошло все фильтры!')import logging
logger = logging.getLogger(__name__)
async def debug_filter(msg: Message) -> bool:
"""
Фильтр для отладки — логирует все сообщения.
Зачем: Понимать, какие сообщения получает бот.
Когда: Отладка проблем с обработкой.
"""
logger.debug(
f"Message from {msg.from_user.username}: {msg.text}"
)
return True
@router.message(debug_filter)
async def debug_handler(msg: Message):
await msg.answer('Сообщение залогировано')💡 Совет: Включите debug-логирование только в development. В production это создаст лишнюю нагрузку.
bot/filters/| Тема | Как связана |
|---|---|
| Архитектура | Фильтры выносятся в bot/filters/, используются в middleware |
| FSM | StateFilter проверяет текущее состояние диалога |
| Безопасность | Фильтры прав доступа (is_admin, is_subscribed) |
| Тестирование | Unit-тесты на кастомные фильтры |
→ Middleware — сквозная логика для всех обработчиков
→ Архитектурные паттерны — модульная структура проекта
→ Тестирование — проверка корректности фильтров
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.