Python Telegram Bot Development — Deep Dive

System-level framing

A production Telegram bot is an async, event-driven service that communicates exclusively through Telegram’s HTTP-based Bot API. The bot never connects directly to users — every message passes through Telegram’s servers. This architecture means your bot must handle API rate limits (30 messages per second globally, 1 per second per chat), network interruptions, and concurrent conversations across thousands of users.

The python-telegram-bot (PTB) library version 20+ is fully async, built on top of asyncio and httpx. It provides the Application class as the central orchestrator.

Project structure

A well-organized bot separates concerns:

telegram-bot/
├── bot/
│   ├── __init__.py
│   ├── main.py          # Application setup and entry point
│   ├── handlers/
│   │   ├── start.py     # /start and /help commands
│   │   ├── settings.py  # User preferences conversation
│   │   └── errors.py    # Global error handler
│   ├── keyboards.py     # Inline keyboard builders
│   ├── persistence.py   # Custom persistence backend
│   └── config.py        # Environment-based configuration
├── tests/
├── Dockerfile
└── pyproject.toml

Application setup with PTB v20+

from telegram.ext import (
    Application, CommandHandler, MessageHandler,
    CallbackQueryHandler, ConversationHandler, filters
)
import os

def main():
    app = (
        Application.builder()
        .token(os.environ["BOT_TOKEN"])
        .persistence(PicklePersistence(filepath="bot_data.pkl"))
        .build()
    )

    # Register handlers
    app.add_handler(CommandHandler("start", start_command))
    app.add_handler(CommandHandler("help", help_command))
    app.add_handler(settings_conversation)
    app.add_handler(CallbackQueryHandler(button_callback))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))

    # Global error handler
    app.add_error_handler(error_handler)

    app.run_polling(allowed_updates=Update.ALL_TYPES)

Handler order matters. PTB checks handlers sequentially and runs the first match, so place specific handlers (commands, conversations) before generic catch-all handlers.

Async handler pattern

Every handler in PTB v20+ is an async function:

from telegram import Update
from telegram.ext import ContextTypes

async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user = update.effective_user
    await update.message.reply_html(
        f"Hello {user.mention_html()}! Send /help to see what I can do."
    )

The context object carries bot data, user data, job queues, and the bot instance itself. Use context.user_data for per-user storage and context.chat_data for per-chat storage.

Conversation handlers in depth

ConversationHandler implements a finite state machine. Each state maps to one or more handlers, and returning a state constant from a handler transitions the conversation:

CHOOSING, TYPING_REPLY = range(2)

settings_conversation = ConversationHandler(
    entry_points=[CommandHandler("settings", settings_start)],
    states={
        CHOOSING: [
            CallbackQueryHandler(choice_handler, pattern="^(language|timezone)$")
        ],
        TYPING_REPLY: [
            MessageHandler(filters.TEXT & ~filters.COMMAND, receive_input)
        ],
    },
    fallbacks=[CommandHandler("cancel", cancel)],
    persistent=True,
    name="settings_conv",
)

Key details:

• persistent=True saves conversation state across bot restarts.
• name is required when persistence is enabled, used as the storage key.
• per_user=True (default) means each user has independent state.
• Conversations time out by default — set conversation_timeout to handle abandoned flows.

Inline keyboards and callback data

from telegram import InlineKeyboardButton, InlineKeyboardMarkup

def build_menu(options: list[str]) -> InlineKeyboardMarkup:
    buttons = [
        [InlineKeyboardButton(text=opt, callback_data=f"select:{opt}")]
        for opt in options
    ]
    return InlineKeyboardMarkup(buttons)

async def button_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()  # Acknowledge the button press

    action, value = query.data.split(":", 1)
    await query.edit_message_text(f"You selected: {value}")

Always call query.answer() to dismiss the loading spinner. Callback data is limited to 64 bytes — for complex payloads, store the data in context.user_data and put only a short key in the callback.

Webhooks for production

Polling works for development but wastes resources in production. Webhooks let Telegram push updates to your server:

app.run_webhook(
    listen="0.0.0.0",
    port=8443,
    url_path=os.environ["BOT_TOKEN"],
    webhook_url=f"https://yourdomain.com/{os.environ['BOT_TOKEN']}",
    cert="cert.pem",  # Self-signed certs are supported
)

Telegram supports ports 443, 80, 88, and 8443 for webhooks. Using the bot token as the URL path prevents unauthorized requests. For most deployments, placing the bot behind a reverse proxy (nginx or Caddy) with a proper TLS certificate is cleaner than self-signed certs.

Rate limiting and flood control

Telegram enforces strict rate limits:

Scope	Limit
Per chat	1 message/second
Per group	20 messages/minute
Global	30 messages/second
Bulk notifications	30 users/second

PTB handles basic queuing, but for broadcast scenarios (sending to thousands of users), implement your own queue with delays:

async def broadcast(context: ContextTypes.DEFAULT_TYPE):
    users = await get_all_subscribers()
    for i, user_id in enumerate(users):
        try:
            await context.bot.send_message(chat_id=user_id, text="Update!")
        except telegram.error.RetryAfter as e:
            await asyncio.sleep(e.retry_after)
            await context.bot.send_message(chat_id=user_id, text="Update!")
        except telegram.error.Forbidden:
            await remove_subscriber(user_id)  # User blocked the bot

        if i % 25 == 0:
            await asyncio.sleep(1)  # Throttle proactively

Persistence backends

PTB ships with PicklePersistence for local storage, but production bots need something more robust:

from telegram.ext import BasePersistence
import redis.asyncio as redis

class RedisPersistence(BasePersistence):
    def __init__(self, url: str):
        super().__init__()
        self.redis = redis.from_url(url)

    async def get_user_data(self) -> dict:
        data = await self.redis.get("user_data")
        return json.loads(data) if data else {}

    async def update_user_data(self, user_id: int, data: dict):
        current = await self.get_user_data()
        current[str(user_id)] = data
        await self.redis.set("user_data", json.dumps(current))
    # ... implement remaining abstract methods

Error handling

A global error handler prevents crashes from killing the bot:

import traceback
import logging

logger = logging.getLogger(__name__)

async def error_handler(update: object, context: ContextTypes.DEFAULT_TYPE):
    logger.error("Exception while handling update:", exc_info=context.error)

    if isinstance(context.error, telegram.error.NetworkError):
        return  # PTB will retry automatically

    if isinstance(context.error, telegram.error.Forbidden):
        # User blocked the bot — clean up
        return

    # Notify admin for unexpected errors
    tb = traceback.format_exception(type(context.error), context.error,
                                     context.error.__traceback__)
    admin_id = int(os.environ["ADMIN_CHAT_ID"])
    await context.bot.send_message(
        chat_id=admin_id,
        text=f"Bot error:\n{''.join(tb)[-3000:]}"
    )

Job queue for scheduled tasks

PTB includes a job queue for deferred and recurring tasks:

async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    # Send a reminder in 30 minutes
    context.job_queue.run_once(
        reminder_callback,
        when=timedelta(minutes=30),
        chat_id=update.effective_chat.id,
        name=f"reminder_{update.effective_user.id}"
    )

    # Daily digest at 9:00 UTC
    context.job_queue.run_daily(
        daily_digest,
        time=time(hour=9, minute=0),
        chat_id=update.effective_chat.id
    )

Deployment patterns

Docker

FROM python:3.12-slim
WORKDIR /app
COPY pyproject.toml .
RUN pip install --no-cache-dir .
COPY bot/ bot/
CMD ["python", "-m", "bot.main"]

Systemd (VPS)

[Unit]
Description=Telegram Bot
After=network.target

[Service]
User=botuser
WorkingDirectory=/opt/telegram-bot
ExecStart=/opt/telegram-bot/.venv/bin/python -m bot.main
Restart=always
RestartSec=5
EnvironmentFile=/opt/telegram-bot/.env

[Install]
WantedBy=multi-user.target

Health monitoring

Add a simple HTTP endpoint alongside the bot for uptime monitoring:

from aiohttp import web

async def health(request):
    return web.json_response({"status": "ok", "uptime": get_uptime()})

# Run alongside the bot
runner = web.AppRunner(web.Application())

Testing bots

PTB provides testing utilities, but the simplest approach is to mock the Telegram API:

import pytest
from unittest.mock import AsyncMock, MagicMock

@pytest.mark.asyncio
async def test_start_command():
    update = MagicMock()
    update.effective_user.mention_html.return_value = "<b>TestUser</b>"
    update.message.reply_html = AsyncMock()
    context = MagicMock()

    await start_command(update, context)

    update.message.reply_html.assert_called_once()
    assert "TestUser" in update.message.reply_html.call_args[0][0]

Tradeoffs: PTB vs aiogram

Aspect	python-telegram-bot	aiogram
Async	v20+ (full async)	Native since v1
Documentation	Extensive, many examples	Good but less English content
Middleware	Via handler groups	Built-in middleware chain
Conversation state	ConversationHandler	FSM with storage backends
Community	Larger, older	Growing, popular in CIS region

Choose PTB for English-first projects with complex conversations. Choose aiogram for projects that need fine-grained middleware control.

One thing to remember: Production Telegram bots are async event-driven services with strict rate limits. Structure your bot around handlers, use persistence for state that survives restarts, and respect Telegram’s throttling to avoid getting temporarily banned.