DEV Community

Steve Jefferson
Steve Jefferson

Posted on Edited on

Rate-limit free vs paid users in a python-telegram-bot bot (one gatekeeper handler)

#python #telegram #tutorial #redis

Most Telegram bots that charge money end up with two kinds of users: free users who get a taste, and paid users who get more. The simplest way to draw that line is a rate limit: free users get, say, 5 requests an hour, paid users get 100.

This post shows how to do it cleanly in python-telegram-bot (v21/v22) with one "gatekeeper" handler that runs before everything else, so you don't sprinkle limit checks through every command.

The idea: a handler that runs first

In python-telegram-bot, handlers live in numbered groups. Groups are processed in ascending order, and within a group only the first matching handler runs. If any handler raises ApplicationHandlerStop, no further handlers run for that update, even in other groups.

So we register a TypeHandler(Update, ...) in group -1. It sees every update before your normal handlers in group 0. If the user is over their limit, it replies once and raises ApplicationHandlerStop. Otherwise it returns and the update flows through as usual.

Step 1: a sliding-window limiter

A sliding window keeps the timestamps of each user's recent requests and counts how many fall inside the window. It's fairer than a fixed "reset on the hour" counter, because a user can't burst 10 requests across 12:59 and 13:00.

# limiter.py
import time
from collections import defaultdict, deque


class SlidingWindowLimiter:
    def __init__(self, window_seconds: int):
        self.window = window_seconds
        self.hits: dict[int, deque[float]] = defaultdict(deque)

    def check(self, user_id: int, limit: int) -> tuple[bool, int]:
        """Return (allowed, seconds_until_next_slot)."""
        now = time.monotonic()
        q = self.hits[user_id]

        # Drop timestamps that have left the window
        while q and now - q[0] >= self.window:
            q.popleft()

        if len(q) < limit:
            q.append(now)
            return True, 0

        retry_after = int(self.window - (now - q[0])) + 1
        return False, retry_after
Enter fullscreen mode Exit fullscreen mode

time.monotonic() is used instead of time.time() so a system clock change can't break the math.

Step 2: decide who is paid

Your source of truth for "paid" might be a database, a license key table, or a webhook from your payment provider. For this tutorial it's a function you can swap out:

# tiers.py
PAID_USER_IDS: set[int] = set()   # fill from your DB on startup

TIERS = {
    "free": 5,     # requests per window
    "paid": 100,
}


def tier_for(user_id: int) -> str:
    return "paid" if user_id in PAID_USER_IDS else "free"
Enter fullscreen mode Exit fullscreen mode

Keep this lookup fast. It runs on every single update, so hitting a remote API here will slow your whole bot down. Cache it in memory and refresh it when a payment event arrives.

Step 3: the gatekeeper handler 

# bot.py
import logging
import os

from telegram import Update
from telegram.ext import (
    Application,
    ApplicationHandlerStop,
    CommandHandler,
    ContextTypes,
    TypeHandler,
)

from limiter import SlidingWindowLimiter
from tiers import TIERS, tier_for

logging.basicConfig(level=logging.INFO)

limiter = SlidingWindowLimiter(window_seconds=3600)


async def gatekeeper(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    user = update.effective_user
    if user is None:          # channel posts, some service updates
        return

    tier = tier_for(user.id)
    allowed, retry_after = limiter.check(user.id, TIERS[tier])
    if allowed:
        return                # let the update continue to group 0

    # Only answer once per blocked burst, so we don't spam the user
    warned = context.user_data.get("rl_warned_until", 0)
    if update.effective_message and retry_after > 0 and warned < retry_after:
        minutes = max(1, retry_after // 60)
        text = (
            f"You've hit the free limit of {TIERS['free']} requests per hour. "
            f"Try again in about {minutes} min, or upgrade with /upgrade."
            if tier == "free"
            else f"Slow down a little, try again in about {minutes} min."
        )
        await update.effective_message.reply_text(text)
    context.user_data["rl_warned_until"] = retry_after

    raise ApplicationHandlerStop


async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    context.user_data.pop("rl_warned_until", None)
    await update.message.reply_text("Hi! Send /price BTC to try me.")


async def price(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    symbol = (context.args[0] if context.args else "BTC").upper()
    # ... call your real API here ...
    await update.message.reply_text(f"{symbol}: (your data here)")


async def upgrade(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text("Link to your checkout page goes here.")


def main() -> None:
    app = Application.builder().token(os.environ["BOT_TOKEN"]).build()

    # Group -1 runs before the default group 0
    app.add_handler(TypeHandler(Update, gatekeeper), group=-1)

    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("price", price))
    app.add_handler(CommandHandler("upgrade", upgrade))

    app.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == "__main__":
    main()
Enter fullscreen mode Exit fullscreen mode

A few details worth calling out:

  • /upgrade is also rate-limited here. You probably want blocked users to still reach it. The easy fix: at the top of gatekeeper, return early if update.effective_message.text starts with /upgrade or /start.
  • The warning is sent once per blocked stretch, not on every message, so a frustrated user mashing buttons doesn't get ten identical replies.
  • ApplicationHandlerStop has no effect in non-blocking handlers. Don't register the gatekeeper with block=False, or the stop is ignored and every update gets through. ## Step 4: survive restarts and multiple workers with Redis

The in-memory limiter resets every time the process restarts and doesn't work if you run more than one instance. Redis fixes both. A sorted set per user gives you the same sliding window:

# redis_limiter.py
import time
import uuid

import redis.asyncio as redis

r = redis.from_url("redis://localhost:6379/0")


async def check(user_id: int, limit: int, window: int = 3600) -> tuple[bool, int]:
    key = f"rl:{user_id}"
    now = time.time()

    async with r.pipeline(transaction=True) as pipe:
        pipe.zremrangebyscore(key, 0, now - window)
        pipe.zcard(key)
        _, count = await pipe.execute()

    if count < limit:
        await r.zadd(key, {f"{now}:{uuid.uuid4().hex}": now})
        await r.expire(key, window)
        return True, 0

    oldest = await r.zrange(key, 0, 0, withscores=True)
    retry_after = int(window - (now - oldest[0][1])) + 1 if oldest else window
    return False, retry_after
Enter fullscreen mode Exit fullscreen mode

Swap limiter.check(...) in the gatekeeper for await check(...) and you're done. The member name includes a random suffix so two requests in the same microsecond don't overwrite each other. There's a small race between the count and the add; for a bot's free tier that's fine. If you need it exact, move the logic into a Lua script so it runs atomically.

Step 5: test it in two minutes

  1. Set TIERS["free"] = 2 temporarily.
  2. Send /price three times. The third should get the limit message and nothing else.
  3. Add your own Telegram user ID to PAID_USER_IDS, restart, and repeat. All three go through.

You can find your user ID by logging update.effective_user.id in any handler.

Things to avoid

  • Don't rate-limit by chat ID in groups. In a group, every member shares one chat ID. Limit by effective_user.id so one noisy member doesn't lock out everyone.
  • Don't confuse this with AIORateLimiter. PTB's built-in AIORateLimiter throttles your bot's outgoing calls to stay under Telegram's flood limits. It doesn't limit what your users can do. You may want both.
  • Don't make the paid check slow. It runs on every update. Cache it.

Wrap-up

One TypeHandler in group -1, a sliding-window counter, and ApplicationHandlerStop give you a clean free/paid split without touching your command handlers. Start in memory, move to Redis when you scale past one process.

I built a kit that packages this (plus templates and a deploy guide): The Python Bot Monetization Kit

Top comments (1)

Collapse
 
sunnatulla_dev profile image
Sunnatulla •

One more update type to let through besides /upgrade: the payment itself. With allowed_updates=Update.ALL_TYPES the gatekeeper also sees pre_checkout_query and the successful_payment message, and both have an effective_user. So a free user who hit the limit, taps /upgrade and pays, gets the pre-checkout dropped by ApplicationHandlerStop. Telegram wants an answer to that within 10 seconds, so the payment just fails, and the user never even gets the limit message because effective_message is None there.

I ran it offline on PTB 22.8 with process_update and a limit of 2: three /price, then a pre_checkout_query, then a successful_payment. Without a bypass the last three all came back blocked. With this at the top of the gatekeeper the payment handlers ran:

if update.pre_checkout_query or (
    update.message and update.message.successful_payment
):
    return
Enter fullscreen mode Exit fullscreen mode

Related small thing: update.effective_message is None for a pre_checkout_query (and for inline queries, poll answers and so on), so the effective_message.text.startswith(...) early return from the post would raise AttributeError there. Checking msg = update.effective_message and msg and msg.text first avoids that.