A telegram bot API client is written in python 3.5+ and currently compatible with Telegram Bot API 5.3 and later. The reason for writing this bot utility is that I wish to run multi telegram bots which could have same or different business logic (route policy) in one process. I reckon it is lightweight, fast, full implement and only urllib3 dependent.
Update for Telegram Bot API 6.1
Update for Telegram Bot API 6.0
Update for Telegram Bot API 5.7 Fix bugs
Update for Telegram Bot API 5.6 using loop.run_in_executor for synchronous functions
fix bugs
fix bugs
fix bugs
fix bugs
- code optimization
- fix bugs
- add group method where methods for InlineKeyboard. more detail is in example.callback_query
- change bot.get_file_bytes(file_path, chunk_size) into bot.get_file_bytes(file_obj)
add remove and where methods for InlineKeyboard. more detail is in example.callback_query
modify router.message_handler(fields) to router.message_handler(*fields)
- code optimization
- get_updates returns a iter
fix a bug for force reply
Remove TelegramRouter from TelegramBot. Do a update dispatch as the update's coming. Have a chance to use yourself route policy on incoming updates.
- change cmds of CommandHandler into *cmds
- change errors of ErrorHandler into *errors
Update for the telegram bot api 5.5. Actually, I did nothing on codes. It is compatible with the telegram bot api 5.5.
- change a list of regex patterns of regex_match into *args of regex patterns
- add a example/regex_match.py for regex_match's usage
Fix bugs
Update for Telegram Bot API 5.4
- HOT FIX remove "del()" from MongoDBStorage
- check and add storage.del() for storages
- rewrite force_reply. see detail in example.force_reply
- rewrite callback_query. see detail in example.callback_query
- add bot.clear_session method
too silly to fix bugs right
fix bugs
- remove UIHelper
- fix bugs
- fix bugs
- optimize codes
- remove ui stack
- fix bugs
- add add_lines in ReplyKeyboard. see example/keyboard.py
- fix bugs
- update ReplyKeyboard in ui. see example/keyboard.py
- update UIHelper in ui. see example/select.py
fix bugs and give the bot a get_file_bytes method to download a file. see example/document.py
optimize codes and remove setup_webhook
Fix bugs on ErrorHandler
- compose keyboards see example/keyboard.py
- fix bugs
Fix bugs
Fix bugs and add router.remove_handler. see example/dynamic_handler.py
-
Refacted and faster than before.
-
Provide a UIHelper in ui for buttons.
A large update. I do not write too many details because no one is using it except myself.
Add: add get_file_url function in bot
Fix bugs and update: support multi emojis for UI buttons
Fix bugs and update: add a UI stack for be back to ahead UI, see example.ui_stack.py
Change: delete multi keys in session using delete function
Add BotCommandScope Support
Fix bugs... correct get_file_bytes in TelegramBotAPI
Optimize: add a context manager on session implement
Fix bugs... correct TelegramBotAPIException
Optimize: make define your local API host easy, and your API host can use 'http://'
Optimize: make all bots call same one TelegramBotAPI instance
Add a confirm in ui
This is a simple echo bot.
from telegrambotclient import bot_client
from telegrambotclient.base import MessageField, ParseMode
# define a default router named "default"
router = bot_client.router()
# decorate a handler callback on incoming message updates that have a text field
@router.message_handler(MessageField.TEXT)
def on_text_message(bot, message):
# receive and reply
sent_message = bot.send_message(
chat_id=message.chat.id,
text="I receive: <strong>{0}</strong>".format(message.text),
parse_mode=ParseMode.HTML,
)
# pin the sent message
bot.pin_chat_message(chat_id=message.chat.id, message_id=sent_message.message_id)
return bot.stop_call
# define a callback for incoming updates
async def on_update(bot, update):
await router.dispatch(bot, update)
# define a bot
bot = bot_client.create_bot(token=<BOT_TOKEN>)
# delete webhook if did or not
bot.delete_webhook(drop_pending_updates=True)
# run polling to fetch updates in every 10s
bot.run_polling(on_update, timeout=10)
telegrambotclient has same parameter signatures with the official Telegram Bot APIs. Please see official Telegram Bot API document when calling telegram bot APIs.
For send_message api, it provides a shortcut.
sent_message = bot.reply_message(
message,
text="I receive: <strong>{0}</strong>".format(message.text),
parse_mode=ParseMode.HTML,
)
In my case, I use fastapi and uvicron to provide a HTTP interface to receive updates from the official Telegram Bot Server. For development and testing, ngrok give a HTTPs URL on my localhost server with a real-time HTTP traffic tunnel.
# run in terminal and get a https tunnel on port 8000 in Austrlia
ngrok http 8000 --region=au
source code:
from fastapi import FastAPI, Request, status
from telegrambotclient import bot_client
from telegrambotclient.base import MessageField, ParseMode
# from ngrok's https url, replace it with yours
WEBHOOK_URL = "https://5f9d0f13b9fb.au.ngrok.io/bot/{0}"
bot_token = "<BOT_TOKEN>"
# define a router
router = bot_client.router()
@router.message_handler(MessageField.TEXT)
def on_text_message(bot, message):
bot.reply_message(message, text="I receive: <strong>{0}</strong> from bot1".format(message.text), parse_mode=ParseMode.HTML)
return bot.stop_call
app = FastAPI()
# waiting for incoming updates and dispatch them
@app.post("/bot/{bot_token}", status_code=status.HTTP_200_OK)
async def process_telegram_update(bot_token: str, request: Request):
bot = bot_client.bots.get(bot_token, None)
if bot:
await router.dispatch(bot, TelegramObject(**await request.json()))
return "OK"
bot = bot_client.create_bot(token=bot_token)
bot.setup_webhook(WEBHOOK_URL.format(bot_token))
from fastapi import FastAPI, Request, status
from telegrambotclient import bot_client
from telegrambotclient.base import Message, MessageField, ParseMode
# from ngrok's https url, replace it with yours
WEBHOOK_URL = "https://5f9d0f13b9fb.au.ngrok.io/bot/{0}"
bot_token_1 = "<BOT_TOKEN_1>"
bot_token_2 = "<BOT_TOKEN_2>"
# define 2 default routers
router1 = bot_client.router(bot_token_1)
router2 = bot_client.router(bot_token_2)
# bind a handler on router1
@router1.message_handler(MessageField.TEXT)
def on_router1_message(bot, message):
bot.reply_message(
message,
text="I receive: <strong>{0}</strong> from router1".format(message.text),
parse_mode=ParseMode.HTML,
)
# bind a handler on router2
@router2.message_handler(MessageField.TEXT)
def on_router2_message(bot, message):
bot.reply_message(
message,
text="I receive: <strong>{0}</strong> from router2".format(message.text),
parse_mode=ParseMode.HTML,
)
app = FastAPI()
# waiting for incoming updates and dispatch them
@app.post("/bot/{bot_token}", status_code=status.HTTP_200_OK)
async def serve_update(bot_token: str, request: Request):
bot = bot_client.bots.get(bot_token, None)
if bot:
router = bot_client.routers.get(bot_token, None)
if router:
await router.dispatch(bot, TelegramObject(**await request.json()))
return "OK"
bot1 = bot_client.create_bot(token=bot_token_1)
bot1.setup_webhook(WEBHOOK_URL.format(bot_token_1))
bot2 = bot_client.create_bot(token=bot_token_2)
bot2.setup_webhook(WEBHOOK_URL.format(bot_token_2))
@router.message_handler(MessageField.TEXT)
def on_message(bot, message):
pass
a good way to register one callback on multi routers
def on_message(bot, message):
pass
router1.register_message_handler(on_message, MessageField.TEXT)
router2.register_message_handler(on_message, MessageField.TEXT)
# callback for a animation message that includes 'animation' AND 'document' fields
@router.message_handler(MessageField.ANIMATION, MessageField.DOCUMENT)
def on_animation(bot, message: Message):
pass