[Python] How to properly rate-limit message sending in aiogram Telegram bot?

I'm building a Telegram bot using aiogram, and I need to limit the rate of sending messages:

• No more than 25 messages per second globally (across all chats).
• No more than 3 messages per 3 seconds per chat.

I'm currently using the limits library with Redis storage backed and have wrapped the Bot class like this:

import asyncio
from typing import Any, Union
from aiogram import Bot
from aiogram.types import Message
from limits import RateLimitItemPerSecond
from limits.aio.strategies import MovingWindowRateLimiter
from limits.storage import storage_from_string

class RateLimitedBot(Bot):
def __init__(
self,
token: str,
redis_url: str,
global_rate: int = 25,
global_per: float = 1.0,
chat_rate: int = 3,
chat_per: float = 3.0,
*args: Any,
**kwargs: Any,
):
super().__init__(token, *args, **kwargs)
self.storage = storage_from_string(redis_url)
self.limiter = MovingWindowRateLimiter(self.storage)

self.global_limit = RateLimitItemPerSecond(global_rate)
self.chat_limit = RateLimitItemPerSecond(chat_rate)

async def wait_for_slot(self, key: str, limit: RateLimitItemPerSecond):
while True:
allowed = await self.limiter.test(limit, key)
if allowed:
await self.limiter.hit(limit, key)
return
await asyncio.sleep(limit.get_expiry() / limit.amount)

async def _apply_rate_limits(self, chat_id: Union[int, str]):
await self.wait_for_slot("rate:global", self.global_limit)
await self.wait_for_slot(f"rate:chat:{chat_id}", self.chat_limit)

async def send_message(
self, chat_id: Union[int, str], text: str, *args: Any, **kwargs: Any
) -> Message:
await self._apply_rate_limits(chat_id)
return await super().send_message(chat_id, text, *args, **kwargs)


The problem is that this implementation sometimes doesn't take the constraints into account, and I don't know why that happens.

Maybe I'm testing the implementation incorrectly, so below is the part where I collect the sending time of each message:

@pytest.mark.asyncio
async def test_rate_limited_bot_does_not_exceed_global_limit():
# bot init

timestamps = []
async def mocked_send_message(*args, **kwargs):
timestamps.append(time.perf_counter())
return "ok"

with patch.object(Bot, "send_message", new=mocked_send_message):
await asyncio.gather(
*[bot.send_message(chat_id=chat_id, text=f"Test {i}") for i in range(message_count)]
)

timestamps = sorted(timestamps)[50:] # skip startup outliers
# check windows


My questions:

• Is there anything obviously wrong in my approach?
• Is there a more reliable or idiomatic way to implement rate limiting in this context?

Continue reading...