Throttling example

throttling_example.py
 1"""
 2Example for throttling manager.
 3
 4You can use that for flood controlling.
 5"""
 6
 7import logging
 8
 9from aiogram import Bot, types
10from aiogram.contrib.fsm_storage.memory import MemoryStorage
11from aiogram.dispatcher import Dispatcher
12from aiogram.utils.exceptions import Throttled
13from aiogram.utils.executor import start_polling
14
15
16API_TOKEN = 'BOT_TOKEN_HERE'
17
18logging.basicConfig(level=logging.INFO)
19
20bot = Bot(token=API_TOKEN)
21
22# Throttling manager does not work without Leaky Bucket.
23# You need to use a storage. For example use simple in-memory storage.
24storage = MemoryStorage()
25dp = Dispatcher(bot, storage=storage)
26
27
28@dp.message_handler(commands=['start'])
29async def send_welcome(message: types.Message):
30    try:
31        # Execute throttling manager with rate-limit equal to 2 seconds for key "start"
32        await dp.throttle('start', rate=2)
33    except Throttled:
34        # If request is throttled, the `Throttled` exception will be raised
35        await message.reply('Too many requests!')
36    else:
37        # Otherwise do something
38        await message.reply("Hi!\nI'm EchoBot!\nPowered by aiogram.")
39
40
41@dp.message_handler(commands=['hi'])
42@dp.throttled(lambda msg, loop, *args, **kwargs: loop.create_task(bot.send_message(msg.from_user.id, "Throttled")),
43              rate=5)
44# loop is added to the function to run coroutines from it
45async def say_hi(message: types.Message):
46    await message.answer("Hi")
47
48
49# the on_throttled object can be either a regular function or coroutine
50async def hello_throttled(*args, **kwargs):
51    # args will be the same as in the original handler
52    # kwargs will be updated with parameters given to .throttled (rate, key, user_id, chat_id)
53    print(f"hello_throttled was called with args={args} and kwargs={kwargs}")
54    message = args[0]  # as message was the first argument in the original handler
55    await message.answer("Throttled")
56
57
58@dp.message_handler(commands=['hello'])
59@dp.throttled(hello_throttled, rate=4)
60async def say_hello(message: types.Message):
61    await message.answer("Hello!")
62
63
64@dp.message_handler(commands=['help'])
65@dp.throttled(rate=5)
66# nothing will happen if the handler will be throttled
67async def help_handler(message: types.Message):
68    await message.answer('Help!')
69
70if __name__ == '__main__':
71    start_polling(dp, skip_updates=True)