Building a Telegram bot in Python, from token to production

import asyncio
from socialift import Telegram

async def main():
    async with Telegram.from_env() as bot:
        me = await bot.get_me()
        print("Bot:", me["username"])

        # chat_id is a numeric ID, or @username for a channel
        await bot.send_message("@my_channel", "<b>Hello!</b> The bot is up.",
                               parse_mode="HTML")

asyncio.run(main())

from socialift import Telegram

async def run_polling():
    async with Telegram.from_env() as bot:
        await bot.delete_webhook()  # getUpdates fails while a webhook is set
        offset = None
        while True:
            updates = await bot.get_updates(offset=offset, timeout=25)
            for update in updates:
                offset = update["update_id"] + 1
                msg = update.get("message") or {}
                if msg.get("text") == "/start":
                    await bot.send_message(msg["chat"]["id"], "Welcome!")

import os
from aiohttp import web
from socialift import Telegram
from socialift.webhooks import verify_telegram_request

SECRET = os.environ["TG_WEBHOOK_SECRET"]  # 1–256 chars: A-Z, a-z, 0-9, _ and -

async def setup(bot: Telegram) -> None:
    await bot.set_webhook("https://example.com/tg/webhook", secret_token=SECRET,
                          allowed_updates=["message", "callback_query"])
    info = await bot.get_webhook_info()
    print(info["pending_update_count"], info.get("last_error_message"))

async def handle(request: web.Request) -> web.Response:
    if not verify_telegram_request(request.headers, SECRET):
        return web.Response(status=403)
    update = await request.json()
    await queue.put(update)  # heavy work goes to a queue; answer fast
    return web.Response(text="ok")

async def send_catalog(bot, chat_id: int) -> None:
    inline = {"inline_keyboard": [[
        {"text": "Catalog", "url": "https://example.com/catalog"},
        {"text": "Place order", "callback_data": "order:42"},
    ]]}
    await bot.send_photo(chat_id, "https://example.com/banner.jpg",
                         caption="New collection", reply_markup=inline)

    menu = {"keyboard": [[{"text": "Contact"}, {"text": "Cart"}]],
            "resize_keyboard": True}
    await bot.send_message(chat_id, "Pick a section:", reply_markup=menu)

async def on_callback(bot, query: dict) -> None:
    await bot.answer_callback_query(query["id"], text="Got it")

import asyncio
import socialift

async def broadcast(bot, chat_ids: list[int], text: str) -> None:
    for chat_id in chat_ids:
        try:
            await bot.send_message(chat_id, text)
        except socialift.RateLimitError as exc:
            await asyncio.sleep(exc.retry_after or 5)
            await bot.send_message(chat_id, text)
        except socialift.PermissionError:
            mark_inactive(chat_id)  # the user blocked the bot (403)
        await asyncio.sleep(0.05)  # ~20 msg/s, under the 30/s ceiling