TL;DR

  • The "hello world" Telegram bot — one loop that reads an update and immediately replies — loses messages on restart, double-sends on retries, and gets rate-limited the first time you broadcast.
  • Split the bot into two loops: a poller that durably records incoming updates, and a sender that delivers outgoing messages from a queue.
  • Use Telegram's update_id as an idempotency key and only advance the getUpdates offset after the update is committed to disk.
  • Store outgoing messages in a SQLite-backed job queue with leases, attempts, and a dead-letter state. You don't need Redis or RabbitMQ for this.
  • Respect 429 Too Many Requests and its retry_after, back off with jitter on network and 5xx errors, and fail fast on permanent errors such as a user blocking the bot.
  • Handle SIGTERM so deploys don't cut messages in half, and export three numbers — queue depth, oldest pending age, dead-letter count — that tell you whether the bot is healthy.

Introduction

Telegram bots are easy to start and surprisingly easy to get wrong in production. The Bot API is simple: call getUpdates (or receive a webhook), call sendMessage. The trouble starts once the bot matters to someone:

  • You deploy a new version and the messages users sent during the restart are never answered — or are answered twice.
  • A reminder or broadcast to a few thousand chats triggers 429 Too Many Requests, and your loop either crashes or hammers the API harder.
  • A user blocks the bot, sendMessage returns 403 Forbidden, and a naive retry loop keeps trying forever.
  • The network blips for 30 seconds and every in-flight message is lost.

This tutorial builds a small, production-oriented bot architecture that handles all of these cases. It uses Python 3.11+, httpx, and SQLite from the standard library. The patterns — durable inbox, outbox queue, leases, idempotency keys, backoff — are the same ones you would use with any messaging API.

Prerequisites

  • Python 3.11 or newer.
  • A bot token from @BotFather.
  • Basic familiarity with asyncio.
  • Docker (optional, for the deployment section).

Install the only third-party dependency:

python -m venv .venv && source .venv/bin/activate pip install "httpx>=0.27"

The architecture 

 Telegram Bot API │ getUpdates (long poll) sendMessage ▼ ▲ ┌──────────┐ INSERT OR IGNORE ┌──────────┐ │ │ poller │ ───────────────────► │ inbox │ │ └──────────┘ (update_id = key) └────┬─────┘ │ │ handler │ ▼ │ ┌──────────┐ claim (lease) ┌──────────┐ │ jobs │ ◄────────────── │ sender │ │ (outbox) │ ──────────────► │ loop │ └──────────┘ done / retry / └──────────┘ dead

The key idea is that nothing important lives only in memory. Incoming updates are written to the inbox table before Telegram is told they were received. Outgoing messages are written to the jobs table before any attempt is made to send them. If the process dies at any point, the next start picks up where the last one stopped.

Step 1 — The schema 

-- schema.sql PRAGMA journal_mode = WAL; CREATE TABLE IF NOT EXISTS inbox ( update_id INTEGER PRIMARY KEY, -- Telegram's id doubles as idempotency key payload TEXT NOT NULL, -- raw JSON update received_at REAL NOT NULL, handled_at REAL -- NULL until the handler has run ); CREATE TABLE IF NOT EXISTS jobs ( id INTEGER PRIMARY KEY AUTOINCREMENT, dedupe_key TEXT UNIQUE, -- e.g. "reply:<update_id>" — prevents double replies chat_id INTEGER NOT NULL, method TEXT NOT NULL DEFAULT 'sendMessage', params TEXT NOT NULL, -- JSON body for the Bot API call status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'running', 'done', 'dead')), attempts INTEGER NOT NULL DEFAULT 0, max_attempts INTEGER NOT NULL DEFAULT 8, run_after REAL NOT NULL, -- unix time; used for backoff and scheduling locked_until REAL, -- lease expiry while running last_error TEXT, created_at REAL NOT NULL, updated_at REAL NOT NULL ); CREATE INDEX IF NOT EXISTS jobs_ready ON jobs (status, run_after); CREATE TABLE IF NOT EXISTS kv ( key TEXT PRIMARY KEY, value TEXT NOT NULL );

Three design choices are worth calling out:

  • update_id as primary key. If the poller receives the same update twice (which happens after a crash between "write" and "acknowledge"), INSERT OR IGNORE makes the second write a no-op.
  • dedupe_key on jobs. The handler that answers update 12345 creates a job with key reply:12345. If the handler runs twice, the second insert is ignored, so the user gets one reply.
  • locked_until lease. A job in running state whose lease has expired is considered abandoned (the process died mid-send) and becomes claimable again.

Step 2 — A small database layer 

# bot/db.py import json import sqlite3 import time from pathlib import Path DB_PATH = Path("bot.db") def connect() -> sqlite3.Connection: conn = sqlite3.connect(DB_PATH, timeout=5.0, isolation_level=None) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=5000") conn.execute("PRAGMA synchronous=NORMAL") return conn def init(conn: sqlite3.Connection) -> None: conn.executescript(Path(__file__).with_name("schema.sql").read_text()) def get_offset(conn) -> int: row = conn.execute("SELECT value FROM kv WHERE key = 'offset'").fetchone() return int(row["value"]) if row else 0 def set_offset(conn, offset: int) -> None: conn.execute( "INSERT INTO kv(key, value) VALUES ('offset', ?) " "ON CONFLICT(key) DO UPDATE SET value = excluded.value", (str(offset),), ) def enqueue(conn, chat_id: int, params: dict, *, dedupe_key: str | None = None, method: str = "sendMessage", delay_s: float = 0.0) -> bool: """Returns True if a new job was created, False if it was a duplicate.""" now = time.time() cur = conn.execute( "INSERT OR IGNORE INTO jobs (dedupe_key, chat_id, method, params, run_after, created_at, updated_at) " "VALUES (?, ?, ?, ?, ?, ?, ?)", (dedupe_key, chat_id, method, json.dumps(params), now + delay_s, now, now), ) return cur.rowcount == 1

We use isolation_level=None (autocommit) and open explicit transactions with BEGIN IMMEDIATE where we need atomicity. IMMEDIATE takes the write lock at the start of the transaction, which avoids lock-upgrade deadlocks between the poller and the sender.

Step 3 — The poller: durable, at-least-once intake

Telegram's getUpdates uses an offset parameter: calling it with offset = last_update_id + 1 tells Telegram that every update up to last_update_id has been received and can be discarded. That gives us a clean rule: write first, acknowledge second.

# bot/poller.py import asyncio import json import logging import time import httpx from bot import db log = logging.getLogger("poller") async def poll_forever(client: httpx.AsyncClient, api: str, stop: asyncio.Event) -> None: conn = db.connect() offset = db.get_offset(conn) backoff = 1.0 while not stop.is_set(): try: resp = await client.get( f"{api}/getUpdates", params={"offset": offset, "timeout": 30, "allowed_updates": json.dumps(["message", "callback_query"])}, timeout=40.0, # must exceed the long-poll timeout  ) data = resp.json() if not data.get("ok"): raise RuntimeError(f"getUpdates failed: {data.get('description')}") updates = data["result"] if updates: now = time.time() conn.execute("BEGIN IMMEDIATE") try: for u in updates: conn.execute( "INSERT OR IGNORE INTO inbox(update_id, payload, received_at) VALUES (?, ?, ?)", (u["update_id"], json.dumps(u), now), ) offset = updates[-1]["update_id"] + 1 db.set_offset(conn, offset) conn.execute("COMMIT") except Exception: conn.execute("ROLLBACK") raise log.info("stored updates", extra={"count": len(updates), "offset": offset}) backoff = 1.0 except (httpx.TransportError, httpx.HTTPStatusError, RuntimeError, ValueError) as exc: log.warning("poll error, backing off", extra={"error": str(exc), "sleep_s": backoff}) try: await asyncio.wait_for(stop.wait(), timeout=backoff) except asyncio.TimeoutError: pass backoff = min(backoff * 2, 60.0) conn.close()

Because the inbox insert and the offset update happen in the same transaction, there are only two possible states after a crash:

  1. The transaction committed: updates are stored and the next call uses the new offset.
  2. It didn't: the next call reuses the old offset, Telegram re-sends the same updates, and INSERT OR IGNORE deduplicates them.

Either way, nothing is lost and nothing is processed twice. This is at-least-once delivery with idempotent processing, which is the practical way to get "exactly once" behavior.

Telegram keeps unacknowledged updates for 24 hours, so even a long outage is recoverable — but only if you don't advance the offset before persisting.

Step 4 — The handler: turn updates into jobs

The handler reads unhandled inbox rows, decides what to do, and enqueues replies. It never calls the Telegram API itself.

# bot/handler.py import asyncio import json import logging import time from bot import db log = logging.getLogger("handler") def build_reply(update: dict) -> tuple[int, dict] | None: msg = update.get("message") if not msg or "text" not in msg: return None chat_id = msg["chat"]["id"] text = msg["text"].strip() if text == "/start": body = "Hi! Send me any text and I'll echo it back." else: body = f"You said: {text[:3500]}" # stay well under the 4096-char limit  return chat_id, {"chat_id": chat_id, "text": body, "reply_parameters": {"message_id": msg["message_id"]}} async def handle_forever(stop: asyncio.Event) -> None: conn = db.connect() while not stop.is_set(): rows = conn.execute( "SELECT update_id, payload FROM inbox WHERE handled_at IS NULL " "ORDER BY update_id LIMIT 100" ).fetchall() for row in rows: update = json.loads(row["payload"]) conn.execute("BEGIN IMMEDIATE") try: reply = build_reply(update) if reply: chat_id, params = reply db.enqueue(conn, chat_id, params, dedupe_key=f"reply:{row['update_id']}") conn.execute("UPDATE inbox SET handled_at = ? WHERE update_id = ?", (time.time(), row["update_id"])) conn.execute("COMMIT") except Exception: conn.execute("ROLLBACK") log.exception("handler failed", extra={"update_id": row["update_id"]}) if not rows: try: await asyncio.wait_for(stop.wait(), timeout=0.5) except asyncio.TimeoutError: pass conn.close()

Enqueuing the reply and marking the update as handled are atomic. If the process dies between them, the transaction rolls back and the update is handled again on restart — and the dedupe_key guarantees only one reply job exists.

In a real bot, build_reply is where your business logic goes: database lookups, calls to an LLM, payment checks. Keeping it free of network calls to Telegram makes it easy to test with plain dictionaries.

If a handler fails on the same update every time (a "poison message"), it will be retried forever. In production, add an attempts column to inbox and skip updates that have failed more than a few times, logging them for manual review.

Step 5 — The sender: leases, retries, and rate limits

The sender is where most of the resilience logic lives. It must:

  • claim one job at a time without two workers taking the same job,
  • classify errors as retryable or permanent,
  • obey retry_after on HTTP 429,
  • stay under Telegram's limits proactively.

Telegram's documented guidance is roughly one message per second per chat, about 20 messages per minute per group, and around 30 messages per second overall for bulk sending. We'll enforce the global and per-chat limits locally so 429s become rare rather than routine.

Claiming a job atomically

SQLite 3.35+ supports UPDATE ... RETURNING, which lets us claim a job in a single statement:

# bot/queue.py import time LEASE_S = 60 CLAIM_SQL = """ UPDATE jobs SET status = 'running', attempts = attempts + 1, locked_until = :lease_until, updated_at = :now WHERE id = ( SELECT id FROM jobs WHERE (status = 'pending' AND run_after <= :now) OR (status = 'running' AND locked_until < :now) -- reclaim abandoned leases ORDER BY run_after LIMIT 1) RETURNING id, chat_id, method, params, attempts, max_attempts """ def claim(conn): now = time.time() return conn.execute(CLAIM_SQL, {"now": now, "lease_until": now + LEASE_S}).fetchone() def complete(conn, job_id: int) -> None: conn.execute("UPDATE jobs SET status='done', locked_until=NULL, updated_at=? WHERE id=?", (time.time(), job_id)) def retry(conn, job_id: int, delay_s: float, error: str) -> None: now = time.time() conn.execute( "UPDATE jobs SET status='pending', run_after=?, locked_until=NULL, last_error=?, updated_at=? WHERE id=?", (now + delay_s, error[:500], now, job_id), ) def bury(conn, job_id: int, error: str) -> None: conn.execute("UPDATE jobs SET status='dead', locked_until=NULL, last_error=?, updated_at=? WHERE id=?", (error[:500], time.time(), job_id))

The lease is what makes this crash-safe. If the process is killed after claiming a job but before marking it done, the job sits in running until locked_until passes, then gets claimed again.

That also means a message can be sent twice in a narrow window: the API call succeeded, but the process died before complete() ran. For chat replies this is usually acceptable. For anything where duplicates are harmful (payments, irreversible actions), make the downstream operation idempotent too — that's a property of the operation, not the queue.

Rate limiting

A simple async limiter that enforces a global rate and a minimum gap per chat:

# bot/ratelimit.py import asyncio import time class Limiter: def __init__(self, global_per_s: float = 25.0, per_chat_gap_s: float = 1.05): self._global_gap = 1.0 / global_per_s self._per_chat_gap = per_chat_gap_s self._next_global = 0.0 self._next_chat: dict[int, float] = {} self._lock = asyncio.Lock() async def wait(self, chat_id: int) -> None: async with self._lock: now = time.monotonic() ready_at = max(now, self._next_global, self._next_chat.get(chat_id, 0.0)) self._next_global = ready_at + self._global_gap self._next_chat[chat_id] = ready_at + self._per_chat_gap if len(self._next_chat) > 10_000: # keep memory bounded  self._next_chat ...
Source：https://dev.to/titanhold/building-a-resilient-telegram-bot-in-python-idempotent-updates-sqlite-job-queue-retries-5ha
My favorites