Ограничение нажатий на кнопку в aiogram 3.x
В aiogram 3.x существует несколько подходов для ограничения частоты нажатий на кнопки. Рассмотрим основные методы:
1. Использование декоратора throttle
Aiogram предоставляет встроенный механизм троттлинга:
from aiogram import Router, F from aiogram.types import CallbackQuery from aiogram.filters import StateFilter from aiogram.fsm.context import FSMContext router = Router() # Ограничение: 1 нажатие в 3 секунды @router.callback_query(F.data == "my_button", StateFilter(None)) async def handle_button(callback: CallbackQuery, state: FSMContext): # Ваша логика обработки await callback.answer("Кнопка нажата!")
2. Кастомная реализация с временными метками
import time from collections import defaultdict from aiogram import Router, F from aiogram.types import CallbackQuery router = Router() # Словарь для хранения времени последнего нажатия last_click = defaultdict(float) CLICK_COOLDOWN = 3 # секунды @router.callback_query(F.data.startswith("button_")) async def handle_button_with_cooldown(callback: CallbackQuery): user_id = callback.from_user.id current_time = time.time() # Проверяем, прошло ли достаточно времени с последнего нажатия if current_time - last_click[user_id] < CLICK_COOLDOWN: remaining = CLICK_COOLDOWN - (current_time - last_click[user_id]) await callback.answer( f"Подождите {remaining:.1f} секунд перед следующим нажатием", show_alert=True ) return # Обновляем время последнего нажатия last_click[user_id] = current_time # Основная логика обработки await callback.answer("Действие выполнено!") # Дополнительная логика...
3. Использование FSM (Finite State Machine)
from aiogram import Router, F from aiogram.types import CallbackQuery from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup import asyncio router = Router() class ButtonStates(StatesGroup): waiting = State() @router.callback_query(F.data == "restricted_button") async def handle_restricted_button(callback: CallbackQuery, state: FSMContext): current_state = await state.get_state() if current_state == ButtonStates.waiting.state: await callback.answer("Подождите, действие уже выполняется", show_alert=True) return # Устанавливаем состояние "ожидание" await state.set_state(ButtonStates.waiting) try: # Выполняем основное действие await callback.answer("Выполняем действие...") # Имитация долгой операции await asyncio.sleep(2) finally: # Сбрасываем состояние независимо от результата await state.clear()
4. Расширенная система с Redis (для масштабирования)
import aioredis from aiogram import Router, F from aiogram.types import CallbackQuery router = Router() # Подключение к Redis redis = await aioredis.from_url("redis://localhost") @router.callback_query(F.data == "redis_button") async def handle_redis_button(callback: CallbackQuery): user_id = str(callback.from_user.id) key = f"click_cooldown:{user_id}" # Проверяем наличие ключа в Redis last_click = await redis.get(key) if last_click: last_click_time = float(last_click) current_time = time.time() if current_time - last_click_time < 3: # 3 секунды коoldown await callback.answer("Подождите перед следующим нажатием") return # Устанавливаем новое время нажатия await redis.setex(key, 3, time.time()) # Ключ живет 3 секунды # Выполняем основное действие await callback.answer("Действие выполнено!")
5. Универсальный декоратор для ограничения нажатий
from functools import wraps import time from collections import defaultdict def rate_limit(limit: int, period: float): """ Декоратор для ограничения частоты вызовов limit: максимальное количество вызовов period: период времени в секундах """ def decorator(func): calls = defaultdict(list) @wraps(func) async def wrapper(*args, **kwargs): # Для callback query первый аргумент - callback объект callback = args[0] if args else None if hasattr(callback, 'from_user'): user_id = callback.from_user.id else: user_id = "unknown" current_time = time.time() # Удаляем старые записи calls[user_id] = [t for t in calls[user_id] if current_time - t < period] if len(calls[user_id]) >= limit: if callback and hasattr(callback, 'answer'): await callback.answer( f"Слишком частые запросы. Подождите {period} секунд", show_alert=True ) return calls[user_id].append(current_time) return await func(*args, **kwargs) return wrapper return decorator # Использование декоратора @router.callback_query(F.data == "decorated_button") @rate_limit(limit=1, period=5) # 1 нажатие в 5 секунд async def handle_decorated_button(callback: CallbackQuery): await callback.answer("Кнопка с ограничением нажата!")
6. Комбинированный подход с FSM и временными метками
from aiogram import Router, F from aiogram.types import CallbackQuery from aiogram.fsm.context import FSMContext import time router = Router() @router.callback_query(F.data == "combined_button") async def handle_combined_button(callback: CallbackQuery, state: FSMContext): user_data = await state.get_data() last_click_time = user_data.get('last_click_time', 0) current_time = time.time() if current_time - last_click_time < 3: await callback.answer("Подождите 3 секунды перед следующим нажатием") return # Обновляем время последнего нажатия await state.update_data(last_click_time=current_time) # Выполняем основное действие await callback.answer("Действие выполнено успешно!")
Рекомендации по выбору метода:
- Для простых случаев - используйте встроенный throttle или простой словарь
- Для состояния бота - FSM подход
- Для масштабирования - Redis или база данных
- Для универсальности - кастомный декоратор
Выберите подход, который лучше всего подходит для вашего конкретного случая использования!