rawapibot.py¶
This example uses only the pure, “bare-metal” API wrapper.
1#!/usr/bin/env python
2"""Simple Bot to reply to Telegram messages.
3
4This is built on the API wrapper, see echobot.py to see the same example built
5on the telegram.ext bot framework.
6This program is dedicated to the public domain under the CC0 license.
7"""
8import asyncio
9import contextlib
10import logging
11from typing import NoReturn
12
13from telegram import Bot, Update
14from telegram.error import Forbidden, NetworkError
15
16logging.basicConfig(
17 format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
18)
19# set higher logging level for httpx to avoid all GET and POST requests being logged
20logging.getLogger("httpx").setLevel(logging.WARNING)
21
22logger = logging.getLogger(__name__)
23
24
25async def main() -> NoReturn:
26 """Run the bot."""
27 # Here we use the `async with` syntax to properly initialize and shutdown resources.
28 async with Bot("TOKEN") as bot:
29 # get the first pending update_id, this is so we can skip over it in case
30 # we get a "Forbidden" exception.
31 try:
32 update_id = (await bot.get_updates())[0].update_id
33 except IndexError:
34 update_id = None
35
36 logger.info("listening for new messages...")
37 while True:
38 try:
39 update_id = await echo(bot, update_id)
40 except NetworkError:
41 await asyncio.sleep(1)
42 except Forbidden:
43 # The user has removed or blocked the bot.
44 update_id += 1
45
46
47async def echo(bot: Bot, update_id: int) -> int:
48 """Echo the message the user sent."""
49 # Request updates after the last update_id
50 updates = await bot.get_updates(offset=update_id, timeout=10, allowed_updates=Update.ALL_TYPES)
51 for update in updates:
52 next_update_id = update.update_id + 1
53
54 # your bot can receive updates without messages
55 # and not all messages contain text
56 if update.message and update.message.text:
57 # Reply to the message
58 logger.info("Found message %s!", update.message.text)
59 await update.message.reply_text(update.message.text)
60 return next_update_id
61 return update_id
62
63
64if __name__ == "__main__":
65 with contextlib.suppress(KeyboardInterrupt): # Ignore exception when Ctrl-C is pressed
66 asyncio.run(main())