Skip to content

Code snippets

青鸟 edited this page Dec 9, 2018 · 182 revisions

This page can be read on its own to find the code snippet you need right now.

It is also a follow-up to the page Introduction to the API. If you come from there, you can leave your command line open and just try out a few of these snippets.

Table of contents generated with markdown-toc

Pure API

Fetch updates

To fetch messages sent to your Bot, you can use the getUpdates API method.

Note: You don't have to use get_updates if you are writing your bot with the telegram.ext submodule, since telegram.ext.Updater takes care of fetching all updates for you. Read more about that here.

>>> updates = bot.get_updates()
>>> print([u.message.text for u in updates])

Fetch images sent to your Bot

>>> updates = bot.get_updates()
>>> print([u.message.photo for u in updates if u.message.photo])

Reply to messages

You'll always need the chat_id

>>> chat_id = bot.get_updates()[-1].message.chat_id

General code snippets

These snippets usually apply to both ways of fetching updates. If you're using telegram.ext, you can get the chat_id in your handler callback with update.message.chat_id.

Note: In general, you can send messages to users by passing their user id as the chat_id. If the bot has a chat with the user, it will send the message to that chat.

Post a text message

ᵀᴱᴸᴱᴳᴿᴬᴹ

>>> bot.send_message(chat_id=chat_id, text="I'm sorry Dave I'm afraid I can't do that.")

Note: send_message method (as any of send_* methods of Bot class) returns the instance of Message class, so it can be used in code later.

Reply to a message

This is a shortcut to bot.send_message with sane defaults. Read more about it in the docs.

>>> update.message.reply_text("I'm sorry Dave I'm afraid I can't do that.")

Note: There are equivalents of this method for replying with photos, audio etc., and similar shortcuts exist throughout the library. Related PRs: #362, #420, #423

Send a chat action

ᵀᴱᴸᴱᴳᴿᴬᴹ Use this to tell the user that something is happening on the bot's side:

>>> bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)

Alternatively, if you have several commands and don't want to repeat the above code snippet inside all commands, you can copy the snippet below and just decorate the callback functions with @send_typing_action.

from functools import wraps

def send_typing_action(func):
    """Sends typing action while processing func command."""

    @wraps(func)
    def command_func(*args, **kwargs):
        bot, update = args
        bot.send_chat_action(chat_id=update.effective_message.chat_id, action=telegram.ChatAction.TYPING)
        return func(bot, update, **kwargs)

    return command_func

@send_typing_action
def my_handler(bot, update):
    pass # Will send 'typing' action while processing the request.

Requesting location and contact from user

>>> location_keyboard = telegram.KeyboardButton(text="send_location", request_location=True)
>>> contact_keyboard = telegram.KeyboardButton(text="send_contact", request_contact=True)
>>> custom_keyboard = [[ location_keyboard, contact_keyboard ]]
>>> reply_markup = telegram.ReplyKeyboardMarkup(custom_keyboard)
>>> bot.send_message(chat_id=chat_id, 
...                  text="Would you mind sharing your location and contact with me?", 
...                  reply_markup=reply_markup)

Message Formatting (bold, italic, code, ...)

Post a text message with Markdown formatting

ᵀᴱᴸᴱᴳᴿᴬᴹ

>>> bot.send_message(chat_id=chat_id, 
...                  text="*bold* _italic_ `fixed width font` [link](http://google.com).", 
...                  parse_mode=telegram.ParseMode.MARKDOWN)

Post a text message with HTML formatting

ᵀᴱᴸᴱᴳᴿᴬᴹ

>>> bot.send_message(chat_id=chat_id, 
...                  text='<b>bold</b> <i>italic</i> <a href="http://google.com">link</a>.', 
...                  parse_mode=telegram.ParseMode.HTML)

Message entities

ᵀᴱᴸᴱᴳᴿᴬᴹ To use MessageEntity, extract the entities from a Message object using parse_entities.

Note: This method should always be used instead of the entities attribute, since it calculates the correct substring from the message text based on UTF-16 codepoints - that is, it extracts the correct string even on when working with weird characters such as Emojis.

>>> entities = message.parse_entities()

There are many more API methods. To read the full API documentation, visit the Telegram API documentation or the library documentation of telegram.Bot

Working with files and media

Post an image file from disk

ᵀᴱᴸᴱᴳᴿᴬᴹ

>>> bot.send_photo(chat_id=chat_id, photo=open('tests/test.png', 'rb'))

Post a voice file from disk

ᵀᴱᴸᴱᴳᴿᴬᴹ

>>> bot.send_voice(chat_id=chat_id, voice=open('tests/telegram.ogg', 'rb'))

Post a photo from a URL

ᵀᴱᴸᴱᴳᴿᴬᴹ

>>> bot.send_photo(chat_id=chat_id, photo='https://telegram.org/img/t_logo.png')

Post a gif from a URL (send_animation)

ᵀᴱᴸᴱᴳᴿᴬᴹ

>>> bot.send_animation(chat_id, animation, duration=None, width=None, height=None, thumb=None, caption=None, parse_mode=None, disable_notification=False, reply_to_message_id=None, reply_markup=None, timeout=20, **kwargs)

See the online documentation

Post an audio from disk

ᵀᴱᴸᴱᴳᴿᴬᴹ

>>> bot.send_audio(chat_id=chat_id, audio=open('tests/test.mp3', 'rb'))

Post a file from disk

ᵀᴱᴸᴱᴳᴿᴬᴹ

>>> bot.send_document(chat_id=chat_id, document=open('tests/test.zip', 'rb'))

Post an image from memory

In this example, image is a PIL (or Pillow) Image object, but it works the same with all media types.

>>> from io import BytesIO
>>> bio = BytesIO()
>>> bio.name = 'image.jpeg'
>>> image.save(bio, 'JPEG')
>>> bio.seek(0)
>>> bot.send_photo(chat_id, photo=bio)

Get image with dimensions closest to a desired size

Where photos is a list of PhotoSize objects and desired_size is a tuple containing the desired size.

>>> def get_closest(photos, desired_size):
>>>     def diff(p): return p.width - desired_size[0], p.height - desired_size[1]
>>>     def norm(t): return abs(t[0] + t[1] * 1j)
>>>     return min(photos, key=lambda p:  norm(diff(p)))

Download a file

ᵀᴱᴸᴱᴳᴿᴬᴹ

>>> file_id = message.voice.file_id
>>> newFile = bot.get_file(file_id)
>>> newFile.download('voice.ogg')

Note: For downloading photos, keep in mind that update.message.photo is an array of different photo sizes. Use update.message.photo[-1] to get the biggest size.

Keyboard Menus

Custom Keyboards

ᵀᴱᴸᴱᴳᴿᴬᴹ

>>> custom_keyboard = [['top-left', 'top-right'], 
...                    ['bottom-left', 'bottom-right']]
>>> reply_markup = telegram.ReplyKeyboardMarkup(custom_keyboard)
>>> bot.send_message(chat_id=chat_id, 
...                  text="Custom Keyboard Test", 
...                  reply_markup=reply_markup)

See also: Build a menu with Buttons

Remove a custom keyboard

>>> reply_markup = telegram.ReplyKeyboardRemove()
>>> bot.send_message(chat_id=chat_id, text="I'm back.", reply_markup=reply_markup)

Other useful stuff

Generate flag emojis from country codes

The Unicode flag emoji for any country can by definition be calculated from the countries 2 letter country code. The following snippet only works in Python 3.

>>> OFFSET = 127462 - ord('A')
>>> 
>>> def flag(code):
...     code = code.upper()
...     return chr(ord(code[0]) + OFFSET) + chr(ord(code[1]) + OFFSET)
... 
>>> flag('de')
'🇩🇪'
>>> flag('us')
'🇺🇸'
>>> flag('ru')
'🇷🇺'
>>>

Get the add group message

class NewMember(BaseFilter):
      def filter(self, message):
          if not message.new_chat_members:
              return False
          return True

def add_group(bot, update):
    for members in update.message.new_chat_members:
        bot.send_message(update.message.chat_id, text="{username} add group".format(username=members.username))

add_group_handle = MessageHandler(callback=add_group, filters=NewMember())

Advanced snippets

Restrict access to a handler (decorator)

This decorator allows you to restrict the access of a handler to only the user_ids specified in LIST_OF_ADMINS.

from functools import wraps

LIST_OF_ADMINS = [12345678, 87654321]

def restricted(func):
    @wraps(func)
    def wrapped(bot, update, *args, **kwargs):
        user_id = update.effective_user.id
        if user_id not in LIST_OF_ADMINS:
            print("Unauthorized access denied for {}.".format(user_id))
            return
        return func(bot, update, *args, **kwargs)
    return wrapped
Usage

Add a @restricted decorator on top of your handler declaration:

@restricted
def my_handler(bot, update):
    pass  # only accessible if `user_id` is in `LIST_OF_ADMINS`.

Send action while handling command (decorator)

This parametrized decorator allows you to signal different actions depending on the type of response of your bot. This way users will have similar feedback from your bot as they would from a real human.

from functools import wraps
from telegram import ChatAction

def send_action(action):
    """Sends `action` while processing func command."""

    def decorator(func):
        @wraps(func)
        def command_func(*args, **kwargs):
            bot, update = args
            bot.send_chat_action(chat_id=update.effective_message.chat_id, action=action)
            return func(bot, update, **kwargs)
        return command_func
    
    return decorator
Usage

Result

You can decorate handler callbacks directly with @send_action(ChatAction.<Action>) or create aliases and decorate with them (more readable) .

send_typing_action = send_action(ChatAction.TYPING)
send_upload_video_action = send_action(ChatAction.UPLOAD_VIDEO)
send_upload_photo_action = send_action(ChatAction.UPLOAD_PHOTO)

With the above aliases, the following decorators are equivalent

@send_typing_action
def my_handler(bot, update):
    pass  # user will see 'typing' while your bot is handling the request.
    
@send_action(ChatAction.TYPING)
def my_handler(bot, update):
    pass  # user will see 'typing' while your bot is handling the request.

All possible actions are documented here.


Build a menu with Buttons

Often times you will find yourself in need for a menu with dynamic content. Use the following build_menu method to create a button layout with n_cols columns out of a list of buttons.

def build_menu(buttons,
               n_cols,
               header_buttons=None,
               footer_buttons=None):
    menu = [buttons[i:i + n_cols] for i in range(0, len(buttons), n_cols)]
    if header_buttons:
        menu.insert(0, header_buttons)
    if footer_buttons:
        menu.append(footer_buttons)
    return menu

You can use the header_buttons and footer_buttons lists to put buttons in the first or last row respectively.

Usage

Output

Replace the ... in below snippet by an appropriate argument, as indicated in the InlineKeyboardButton documentation. If you want to use KeyboardButtons, use ReplyKeyboardMarkup instead of InlineKeyboardMarkup.

button_list = [
    InlineKeyboardButton("col1", callback_data=...),
    InlineKeyboardButton("col2", callback_data=...),
    InlineKeyboardButton("row 2", callback_data=...)
]
reply_markup = InlineKeyboardMarkup(util.build_menu(button_list, n_cols=2))
bot.send_message(..., "A two-column menu", reply_markup=reply_markup)

Or, if you need a dynamic version, use list comprehension to generate your button_list dynamically from a list of strings:

some_strings = ["col1", "col2", "row2"]
button_list = [[KeyboardButton(s)] for s in some_strings]

This is especially useful if put inside a helper method like get_data_buttons to work on dynamic data and updating the menu according to user input.

To handle the callback_data, you need to set a CallbackQueryHandler.

Cached Telegram group administrator check

If you want to limit certain bot functions to group administrators, you have to test if a user is an administrator in the group in question. This however requires an extra API request, which is why it can make sense to cache this information for a certain time, especially if your bot is very busy.

This snippet requires this timeout-based cache decorator. (gist mirror)

Save the decorator to a new file named mwt.py and add this line to your imports:

from mwt import MWT

Then, add the following decorated function to your script. You can change the timeout as required.

@MWT(timeout=60*60)
def get_admin_ids(bot, chat_id):
    """Returns a list of admin IDs for a given chat. Results are cached for 1 hour."""
    return [admin.user.id for admin in bot.get_chat_administrators(chat_id)]

You can then use the function like this:

if update.message.from_user.id in get_admin_ids(bot, update.message.chat_id):
    # admin only

Note: Private chats and groups with all_members_are_administrator flag, are not covered by this snippet. Make sure you handle them.

Simple way of restarting the bot

The following example allows you to restart the bot from within a handler. It goes without saying that you should protect this method from access by unauthorized users, which is why we are using a Filters.user filter. If you want multiple users to have access the restart command, you can pass a list of usernames as well. You can also filter by user IDs which is arguably a bit safer since they can't change. See the docs for more information.

This example is using closures so it has access to the updater variable. Alternatively, you could make it global.

import os
import sys
from threading import Thread

# Other code

def main():
    updater = Updater("TOKEN")
    dp = updater.dispatcher

    # Add your other handlers here...

    def stop_and_restart():
        """Gracefully stop the Updater and replace the current process with a new one"""
        updater.stop()
        os.execl(sys.executable, sys.executable, *sys.argv)

    def restart(bot, update):
        update.message.reply_text('Bot is restarting...')
        Thread(target=stop_and_restart).start()

    # ...or here...

    dp.add_handler(CommandHandler('r', restart, filters=Filters.user(username='@jh0ker')))

    # ...or here, depending on your preference :)

    updater.start_polling()
    updater.idle()


if __name__ == '__main__':
    main()

Store ConversationHandler States

The following code allows you to store ConversationHandler States and UserData and reloading them when you restart the bot. Store procedure is executed every 60 seconds; to change this value, you can modify the time.sleep(60) instruction.

You should declare the two methods at the end of the main method to use python closure for accessing ConversationHandler and UserData.

import time, threading, pickle

def main():
    def loadData():
        try:
            f = open('backup/conversations', 'rb')
            conv_handler.conversations = pickle.load(f)
            f.close()
            f = open('backup/userdata', 'rb')
            dp.user_data = pickle.load(f)
            f.close()
        except FileNotFoundError:
            utils.logging.error("Data file not found")         
        except:
            utils.logging.error(sys.exc_info()[0])         
 
    def saveData():
        while True:
            time.sleep(60)
            # Before pickling
            resolved = dict()
            for k, v in conv_handler.conversations.items():
                if isinstance(v, tuple) and len(v) is 2 and isinstance(v[1], Promise):
                    try:
                        new_state = v[1].result()  # Result of async function
                    except:
                        new_state = v[0]  # In case async function raised an error, fallback to old state
                    resolved[k] = new_state
                else:
                    resolved[k] = v
            try:
                f = open('backup/conversations', 'wb+')
                pickle.dump(resolved, f)
                f.close()
                f = open('backup/userdata', 'wb+')
                pickle.dump(dp.user_data, f)
                f.close()
            except:
                utils.logging.error(sys.exc_info()[0])
Usage
def main():
     ...
     loadData()
     threading.Thread(target=saveData).start()

Save and load jobs using pickle

The following snippet pickles the jobs in the job queue periodically and on bot shutdown and unpickles and queues them again on startup. Since pickle doesn't support threading primitives, they are converted.

Note: Race condition for asynchronous jobs that use job.job_queue, job.removed, job.schedule_removal or job.enabled while the job is being pickled.

import pickle
from threading import Event
from time import time
from datetime import timedelta


JOBS_PICKLE = 'job_tuples.pickle'


def load_jobs(jq):
    now = time()

    with open(JOBS_PICKLE, 'rb') as fp:
        while True:
            try:
                next_t, job = pickle.load(fp)
            except EOFError:
                break  # Loaded all job tuples

            # Create threading primitives
            enabled = job._enabled
            removed = job._remove

            job._enabled = Event()
            job._remove = Event()

            if enabled:
                job._enabled.set()

            if removed:
                job._remove.set()

            next_t -= now  # Convert from absolute to relative time

            jq._put(job, next_t)


def save_jobs(jq):
    if jq:
        job_tuples = jq._queue.queue
    else:
        job_tuples = []

    with open(JOBS_PICKLE, 'wb') as fp:
        for next_t, job in job_tuples:
            # Back up objects
            _job_queue = job._job_queue
            _remove = job._remove
            _enabled = job._enabled

            # Replace un-pickleable threading primitives
            job._job_queue = None  # Will be reset in jq.put
            job._remove = job.removed  # Convert to boolean
            job._enabled = job.enabled  # Convert to boolean

            # Pickle the job
            pickle.dump((next_t, job), fp)

            # Restore objects
            job._job_queue = _job_queue
            job._remove = _remove
            job._enabled = _enabled


def save_jobs_job(bot, job):
    save_jobs(job.job_queue)


def main():
    # updater = Updater(..)

    job_queue = updater.job_queue

    # Periodically save jobs
    job_queue.run_repeating(save_jobs_job, timedelta(minutes=1))

    try:
        load_jobs(job_queue)

    except FileNotFoundError:
        # First run
        pass

    # updater.start_[polling|webhook]()
    # updater.idle()

    # Run again after bot has been properly shut down
    save_jobs(job_queue)

if __name__ == '__main__':
    main()

What to read next?

If you haven't read the tutorial "Extensions – Your first Bot" yet, you might want to do it now.

Clone this wiki locally