Building a Resilient Telegram Bot in Python: Idempotent Updates, SQLite Job Queue, Retries, Graceful Shutdown

Hennadii Kovalenko — Sat, 31 Oct 2026 09:07:18 +0000

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:

The transaction committed: updates are stored and the next call uses the new offset.
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 = {c: t for c, t in self._next_chat.items() if t > now}
        delay = ready_at - time.monotonic()
        if delay > 0:
            await asyncio.sleep(delay)

We deliberately run below the documented ceiling (25/s instead of 30/s) to leave headroom for the poller and any other API calls.

Classifying errors

# bot/errors.py
import random

PERMANENT_CODES = {400, 401, 403, 404}   # bad request, bad token, blocked/kicked, not found

def backoff_delay(attempt: int, base: float = 2.0, cap: float = 600.0) -> float:
    """Exponential backoff with full jitter."""
    return random.uniform(0, min(cap, base * (2 ** (attempt - 1))))

429 â€” always retry, after exactly parameters.retry_after seconds. Don't add your own exponential growth on top; Telegram tells you how long to wait.
5xx and network errors â€” retry with exponential backoff and jitter. Jitter prevents every queued job from retrying at the same instant after an outage.
400 / 403 / 404 â€” permanent. A 403 Forbidden: bot was blocked by the user will never succeed by retrying. Move the job to dead and, ideally, mark the chat as inactive so you stop enqueuing messages for it.
401 â€” your token is invalid. This is not a per-job problem; the whole sender should stop and alert.

The sender loop

# bot/sender.py
import asyncio
import json
import logging
import httpx
from bot import db, queue
from bot.errors import PERMANENT_CODES, backoff_delay
from bot.ratelimit import Limiter

log = logging.getLogger("sender")

async def send_forever(client: httpx.AsyncClient, api: str, stop: asyncio.Event) -> None:
    conn = db.connect()
    limiter = Limiter()
    while not stop.is_set():
        job = queue.claim(conn)
        if job is None:
            try:
                await asyncio.wait_for(stop.wait(), timeout=0.5)
            except asyncio.TimeoutError:
                pass
            continue

        await limiter.wait(job["chat_id"])
        ctx = {"job_id": job["id"], "chat_id": job["chat_id"], "attempt": job["attempts"]}
        try:
            resp = await client.post(f"{api}/{job['method']}",
                                     json=json.loads(job["params"]), timeout=15.0)
            body = resp.json()
        except (httpx.TransportError, ValueError) as exc:
            _retry_or_bury(conn, job, backoff_delay(job["attempts"]), f"network: {exc}", ctx)
            continue

        if body.get("ok"):
            queue.complete(conn, job["id"])
            log.info("sent", extra=ctx)
            continue

        code = body.get("error_code", resp.status_code)
        desc = body.get("description", "")
        if code == 429:
            wait_s = float(body.get("parameters", {}).get("retry_after", 5))
            # Retry-after is not a failure of this job: give the attempt back.
            conn.execute("UPDATE jobs SET attempts = attempts - 1 WHERE id = ?", (job["id"],))
            queue.retry(conn, job["id"], wait_s, f"429: {desc}")
            log.warning("rate limited", extra={**ctx, "retry_after": wait_s})
            await asyncio.sleep(wait_s)          # pause the whole sender, not just this job
        elif code == 401:
            queue.retry(conn, job["id"], 60, f"401: {desc}")
            log.critical("bot token rejected; stopping sender", extra=ctx)
            stop.set()
        elif code in PERMANENT_CODES:
            queue.bury(conn, job["id"], f"{code}: {desc}")
            log.warning("permanent failure", extra={**ctx, "code": code, "desc": desc})
        else:
            _retry_or_bury(conn, job, backoff_delay(job["attempts"]), f"{code}: {desc}", ctx)
    conn.close()

def _retry_or_bury(conn, job, delay: float, error: str, ctx: dict) -> None:
    if job["attempts"] >= job["max_attempts"]:
        queue.bury(conn, job["id"], error)
        log.error("giving up", extra={**ctx, "error": error})
    else:
        queue.retry(conn, job["id"], delay, error)
        log.warning("retrying", extra={**ctx, "error": error, "delay_s": round(delay, 1)})

A 429 is a signal about the whole bot, not just one chat, so the sender sleeps for retry_after before claiming anything else. Without that pause, a burst of queued jobs would each hit 429 in turn and extend the penalty.

Step 6 â€” Wiring it together with graceful shutdown

# bot/main.py
import asyncio
import logging
import os
import signal
import httpx
from bot import db
from bot.poller import poll_forever
from bot.handler import handle_forever
from bot.sender import send_forever

async def main() -> None:
    logging.basicConfig(level=logging.INFO,
                        format="%(asctime)s %(levelname)s %(name)s %(message)s")
    token = os.environ["TELEGRAM_BOT_TOKEN"]
    api = f"https://api.telegram.org/bot{token}"

    conn = db.connect()
    db.init(conn)
    conn.close()

    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, stop.set)

    limits = httpx.Limits(max_connections=20, max_keepalive_connections=10)
    async with httpx.AsyncClient(limits=limits) as client:
        tasks = [
            asyncio.create_task(poll_forever(client, api, stop), name="poller"),
            asyncio.create_task(handle_forever(stop), name="handler"),
            asyncio.create_task(send_forever(client, api, stop), name="sender"),
        ]
        await stop.wait()
        logging.info("shutdown: waiting for loops to finish current work")
        done, pending = await asyncio.wait(tasks, timeout=20)
        for t in pending:
            t.cancel()

if __name__ == "__main__":
    asyncio.run(main())

Each loop checks stop between units of work, so on SIGTERM the sender finishes the message it is currently sending, the handler finishes the current transaction, and the poller exits after its current long-poll returns (up to 30 seconds â€” which is why the shutdown waits 20 seconds and then cancels; a cancelled long-poll is harmless because the offset was never advanced for it).

Never log the full api URL: it contains your bot token. If you log HTTP errors from httpx, scrub the token from the message first.

Step 7 â€” Deploying it

A minimal container setup:

# Dockerfile
FROM python:3.12-slim
WORKDIR /app
RUN pip install --no-cache-dir "httpx>=0.27"
COPY bot/ ./bot/
ENV PYTHONUNBUFFERED=1
WORKDIR /data
CMD ["python", "-m", "bot.main"]

# compose.yaml
services:
  bot:
    build: .
    environment:
      TELEGRAM_BOT_TOKEN: ${TELEGRAM_BOT_TOKEN}
      PYTHONPATH: /app
    volumes:
      - botdata:/data          # bot.db lives here and survives redeploys
    restart: unless-stopped
    stop_grace_period: 30s     # more than the 20s internal shutdown budget
    logging:
      driver: json-file
      options: { max-size: "10m", max-file: "5" }
volumes:
  botdata:

Run exactly one instance when using getUpdates. Two pollers with the same token will fight over the offset, and Telegram returns 409 Conflict to one of them. If you need horizontal scaling, switch the intake to a webhook (the inbox pattern stays the same: the webhook handler does INSERT OR IGNORE and returns 200 immediately) and run multiple senders â€” the lease-based claim already supports that.

Step 8 â€” Observability: three numbers that matter

You can tell almost everything about this bot's health from the jobs table:

# bot/stats.py
import time
from bot import db

def snapshot() -> dict:
    conn = db.connect()
    now = time.time()
    try:
        pending = conn.execute(
            "SELECT COUNT(*) AS n, MIN(created_at) AS oldest FROM jobs WHERE status='pending'"
        ).fetchone()
        dead_24h = conn.execute(
            "SELECT COUNT(*) AS n FROM jobs WHERE status='dead' AND updated_at > ?", (now - 86400,)
        ).fetchone()
        unhandled = conn.execute(
            "SELECT COUNT(*) AS n FROM inbox WHERE handled_at IS NULL"
        ).fetchone()
    finally:
        conn.close()
    return {
        "queue_depth": pending["n"],
        "oldest_pending_age_s": round(now - pending["oldest"]) if pending["oldest"] else 0,
        "dead_last_24h": dead_24h["n"],
        "inbox_unhandled": unhandled["n"],
    }

Oldest pending age is the single best alerting signal. Queue depth can legitimately spike during a broadcast; but if the oldest message has been waiting for 10 minutes, users are waiting too.
Dead jobs per day tells you about permanent failures. A steady trickle of 403 is normal (people block bots). A sudden spike of 400 usually means a code change produced invalid requests â€” for example, malformed parse_mode markup.
Inbox unhandled growing means the handler is stuck or crashing.

Expose snapshot() through a tiny HTTP endpoint, log it once a minute as a structured line, or push it to your metrics system. Pair it with a heartbeat: have the sender ping an external heartbeat URL once a minute, so a dead process triggers an alert even when nothing is queued.

Housekeeping matters too. Delete done jobs and handled inbox rows older than a few days so the database stays small:

DELETE FROM jobs  WHERE status = 'done' AND updated_at < unixepoch() - 3 * 86400;
DELETE FROM inbox WHERE handled_at IS NOT NULL AND handled_at < unixepoch() - 3 * 86400;

(unixepoch() requires SQLite 3.38+; on older versions use strftime('%s','now').)

Step 9 â€” Testing the failure paths

The happy path is easy to test by sending the bot a message. The failure paths need deliberate tests. Because the sender only talks to an HTTP API, httpx.MockTransport lets you simulate Telegram's responses without the network:

# tests/test_sender.py
import asyncio
import json
import httpx
import pytest
from bot import db, queue
from bot.sender import send_forever

@pytest.fixture
def fresh_db(tmp_path, monkeypatch):
    monkeypatch.setattr(db, "DB_PATH", tmp_path / "test.db")
    conn = db.connect(); db.init(conn)
    yield conn
    conn.close()

def run_sender(handler, seconds=1.5):
    async def go():
        stop = asyncio.Event()
        transport = httpx.MockTransport(handler)
        async with httpx.AsyncClient(transport=transport) as client:
            task = asyncio.create_task(send_forever(client, "https://api.test/botX", stop))
            await asyncio.sleep(seconds)
            stop.set()
            await task
    asyncio.run(go())

def test_blocked_user_goes_to_dead_letter(fresh_db):
    db.enqueue(fresh_db, 42, {"chat_id": 42, "text": "hi"})
    run_sender(lambda req: httpx.Response(403, json={
        "ok": False, "error_code": 403, "description": "Forbidden: bot was blocked by the user"}))
    row = fresh_db.execute("SELECT status, attempts FROM jobs").fetchone()
    assert row["status"] == "dead" and row["attempts"] == 1

def test_429_is_retried_without_burning_attempts(fresh_db):
    db.enqueue(fresh_db, 42, {"chat_id": 42, "text": "hi"})
    calls = {"n": 0}
    def handler(req):
        calls["n"] += 1
        if calls["n"] == 1:
            return httpx.Response(429, json={"ok": False, "error_code": 429,
                "description": "Too Many Requests: retry after 1", "parameters": {"retry_after": 1}})
        return httpx.Response(200, json={"ok": True, "result": {}})
    run_sender(handler, seconds=3)
    row = fresh_db.execute("SELECT status, attempts FROM jobs").fetchone()
    assert row["status"] == "done" and row["attempts"] == 1

Also worth testing:

killing the process with kill -9 mid-broadcast, then restarting and checking that every job ends in done or dead;
sending the same update payload to the handler twice and asserting a single job;
disconnecting the network for a minute and confirming the poller backs off and recovers.

Trade-offs and when to outgrow this

SQLite is a single-writer database. A bot doing hundreds of sends per second, or running across several machines, should move the queue to PostgreSQL (with SELECT ... FOR UPDATE SKIP LOCKED) or a broker. The schema and state machine in this tutorial carry over almost unchanged.
Long polling vs. webhooks. Long polling is simpler to run behind NAT and needs no TLS endpoint; webhooks scale better and reduce latency. The inbox/outbox design works with both.
Ordering. The sender processes jobs by run_after, so retries can reorder messages within a chat. If strict per-chat ordering matters, claim only the oldest pending job per chat and skip chats that already have a job running.

Conclusion

A reliable Telegram bot is mostly about where state lives. By persisting updates before acknowledging them, persisting replies before sending them, and treating every API call as something that can fail, you get a bot that survives restarts, network drops, rate limits, and blocked users â€” with nothing more than Python, httpx, and a SQLite file.

The same structure â€” durable inbox, idempotent handler, leased outbox with classified retries â€” applies to WhatsApp, Slack, Discord, and email sending. Once you've built it once, you'll recognize it everywhere.

About the author

Hennadii Kovalenko is a software engineer who has been programming since 2004. He builds and operates Telegram bots, backend services, and AI-agent tooling, and runs TITANHOLD, a small software studio. He writes about making small production systems boringly reliable.

DEV Community: Hennadii Kovalenko