Introduction

amanobot reference

Amanobot has two versions:

  • Traditional version works on Python 2.7 and Python 3. It uses urllib3 to make HTTP requests, and uses threads to achieve delegation by default.
  • Async version works on Python 3.5 or above. It is based on asyncio, uses aiohttp to make asynchronous HTTP requests, and uses asyncio tasks to achieve delegation.

This page focuses on traditional version. Async version is very similar, the most significant differences being:

  • Blocking methods (mostly network operations) become coroutines, and should be called with await.
  • Delegation is achieved by tasks, instead of threads. Thread-safety ceases to be a concern.

Traditional modules are under the package amanobot, while async modules are under amanobot.aio:

Traditional Async
amanobot amanobot.aio
amanobot.loop amanobot.aio.loop
amanobot.delegate amanobot.aio.delegate
amanobot.helper amanobot.aio.helper
amanobot.routing amanobot.aio.routing
amanobot.api amanobot.aio.api

Some modules do not have async counterparts, e.g. amanobot.namedtuple and amanobot.exception, because they are shared.

Try to combine this reading with the provided examples . One example is worth a thousand words. I hope they make things clear.

Basic Bot

The Bot class is mostly a wrapper around Telegram Bot API. Many methods are straight mappings to Bot API methods. Where appropriate, I only give links below. No point to duplicate all the details.

class amanobot.Bot(token, raise_errors=True)[source]
class Scheduler[source]
class Event(timestamp, data)

Create new instance of Event(timestamp, data)

data

Alias for field number 1

timestamp

Alias for field number 0

event_at(when, data)[source]

Schedule some data to emit at an absolute timestamp.

Returns:an internal Event object
event_later(delay, data)[source]

Schedule some data to emit after a number of seconds.

Returns:an internal Event object
event_now(data)[source]

Emit some data as soon as possible.

Returns:an internal Event object
cancel(event)[source]

Cancel an event.

run()[source]

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

getMe()[source]

See: https://core.telegram.org/bots/api#getme

sendMessage(chat_id, text, parse_mode=None, disable_web_page_preview=None, disable_notification=None, reply_to_message_id=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#sendmessage

forwardMessage(chat_id, from_chat_id, message_id, disable_notification=None)[source]

See: https://core.telegram.org/bots/api#forwardmessage

sendPhoto(chat_id, photo, caption=None, parse_mode=None, disable_notification=None, reply_to_message_id=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#sendphoto

Parameters:photo –
  • string: file_id for a photo existing on Telegram servers
  • string: HTTP URL of a photo from the Internet
  • file-like object: obtained by open(path, 'rb')
  • tuple: (filename, file-like object). If the filename contains non-ASCII characters and you are using Python 2.7, make sure the filename is a unicode string.
sendAudio(chat_id, audio, caption=None, parse_mode=None, duration=None, performer=None, title=None, thumb=None, disable_notification=None, reply_to_message_id=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#sendaudio

Parameters:audio – Same as photo in amanobot.Bot.sendPhoto()
sendDocument(chat_id, document, thumb=None, caption=None, parse_mode=None, disable_notification=None, reply_to_message_id=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#senddocument

Parameters:document – Same as photo in amanobot.Bot.sendPhoto()
sendVideo(chat_id, video, duration=None, width=None, height=None, thumb=None, caption=None, parse_mode=None, supports_streaming=None, disable_notification=None, reply_to_message_id=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#sendvideo

Parameters:video – Same as photo in amanobot.Bot.sendPhoto()
sendAnimation(chat_id, animation, duration=None, width=None, height=None, thumb=None, caption=None, parse_mode=None, disable_notification=None, reply_to_message_id=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#sendanimation

Parameters:animation – Same as photo in amanobot.Bot.sendPhoto()
sendVoice(chat_id, voice, caption=None, parse_mode=None, duration=None, disable_notification=None, reply_to_message_id=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#sendvoice

Parameters:voice – Same as photo in amanobot.Bot.sendPhoto()
sendVideoNote(chat_id, video_note, duration=None, length=None, thumb=None, disable_notification=None, reply_to_message_id=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#sendvideonote

Parameters:
  • video_note – Same as photo in amanobot.Bot.sendPhoto()
  • length – Although marked as optional, this method does not seem to work without it being specified. Supply any integer you want. It seems to have no effect on the video note’s display size.
sendMediaGroup(chat_id, media, disable_notification=None, reply_to_message_id=None)[source]

See: https://core.telegram.org/bots/api#sendmediagroup

Parameters:media (array of InputMedia objects) –

To indicate media locations, each InputMedia object’s media field should be one of these:

  • string: file_id for a file existing on Telegram servers
  • string: HTTP URL of a file from the Internet
  • file-like object: obtained by open(path, 'rb')
  • tuple: (form-data name, file-like object)
  • tuple: (form-data name, (filename, file-like object))

In case of uploading, you may supply customized multipart/form-data names for each uploaded file (as in last 2 options above). Otherwise, amanobot assigns unique names to each uploaded file. Names assigned by amanobot will not collide with user-supplied names, if any.
sendLocation(chat_id, latitude, longitude, live_period=None, disable_notification=None, reply_to_message_id=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#sendlocation

editMessageLiveLocation(msg_identifier, latitude, longitude, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#editmessagelivelocation

Parameters:msg_identifier – Same as in Bot.editMessageText()
stopMessageLiveLocation(msg_identifier, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#stopmessagelivelocation

Parameters:msg_identifier – Same as in Bot.editMessageText()
sendVenue(chat_id, latitude, longitude, title, address, foursquare_id=None, foursquare_type=None, disable_notification=None, reply_to_message_id=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#sendvenue

sendContact(chat_id, phone_number, first_name, last_name=None, vcard=None, disable_notification=None, reply_to_message_id=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#sendcontact

sendPoll(chat_id, question, options, is_anonymous=None, type=None, allows_multiple_answers=None, correct_option_id=None, explanation=None, explanation_parse_mode=None, open_period=None, is_closed=None, disable_notification=None, reply_to_message_id=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#sendpoll

sendDice(chat_id, emoji=None, disable_notification=None, reply_to_message_id=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#senddice

sendGame(chat_id, game_short_name, disable_notification=None, reply_to_message_id=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#sendgame

sendInvoice(chat_id, title, description, payload, provider_token, start_parameter, currency, prices, provider_data=None, photo_url=None, photo_size=None, photo_width=None, photo_height=None, need_name=None, need_phone_number=None, need_email=None, need_shipping_address=None, is_flexible=None, disable_notification=None, reply_to_message_id=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#sendinvoice

sendChatAction(chat_id, action)[source]

See: https://core.telegram.org/bots/api#sendchataction

getUserProfilePhotos(user_id, offset=None, limit=None)[source]

See: https://core.telegram.org/bots/api#getuserprofilephotos

getFile(file_id)[source]

See: https://core.telegram.org/bots/api#getfile

kickChatMember(chat_id, user_id, until_date=None)[source]

See: https://core.telegram.org/bots/api#kickchatmember

unbanChatMember(chat_id, user_id)[source]

See: https://core.telegram.org/bots/api#unbanchatmember

restrictChatMember(chat_id, user_id, until_date=None, can_send_messages=None, can_send_media_messages=None, can_send_polls=None, can_send_other_messages=None, can_add_web_page_previews=None, can_change_info=None, can_invite_users=None, can_pin_messages=None, permissions=None)[source]

See: https://core.telegram.org/bots/api#restrictchatmember

promoteChatMember(chat_id, user_id, can_change_info=None, can_post_messages=None, can_edit_messages=None, can_delete_messages=None, can_invite_users=None, can_restrict_members=None, can_pin_messages=None, can_promote_members=None)[source]

See: https://core.telegram.org/bots/api#promotechatmember

setChatAdministratorCustomTitle(chat_id, user_id, custom_title)[source]

See: https://core.telegram.org/bots/api#setchatadministratorcustomtitle

setChatPermissions(chat_id, can_send_messages=None, can_send_media_messages=None, can_send_polls=None, can_send_other_messages=None, can_add_web_page_previews=None, can_change_info=None, can_invite_users=None, can_pin_messages=None, permissions=None)[source]

See: https://core.telegram.org/bots/api#setchatpermissions

See: https://core.telegram.org/bots/api#exportchatinvitelink

setChatPhoto(chat_id, photo)[source]

See: https://core.telegram.org/bots/api#setchatphoto

deleteChatPhoto(chat_id)[source]

See: https://core.telegram.org/bots/api#deletechatphoto

setChatTitle(chat_id, title)[source]

See: https://core.telegram.org/bots/api#setchattitle

setChatDescription(chat_id, description=None)[source]

See: https://core.telegram.org/bots/api#setchatdescription

pinChatMessage(chat_id, message_id, disable_notification=None)[source]

See: https://core.telegram.org/bots/api#pinchatmessage

unpinChatMessage(chat_id)[source]

See: https://core.telegram.org/bots/api#unpinchatmessage

leaveChat(chat_id)[source]

See: https://core.telegram.org/bots/api#leavechat

getChat(chat_id)[source]

See: https://core.telegram.org/bots/api#getchat

getChatAdministrators(chat_id)[source]

See: https://core.telegram.org/bots/api#getchatadministrators

getChatMembersCount(chat_id)[source]

See: https://core.telegram.org/bots/api#getchatmemberscount

getChatMember(chat_id, user_id)[source]

See: https://core.telegram.org/bots/api#getchatmember

setChatStickerSet(chat_id, sticker_set_name)[source]

See: https://core.telegram.org/bots/api#setchatstickerset

deleteChatStickerSet(chat_id)[source]

See: https://core.telegram.org/bots/api#deletechatstickerset

answerCallbackQuery(callback_query_id, text=None, show_alert=None, url=None, cache_time=None)[source]

See: https://core.telegram.org/bots/api#answercallbackquery

setMyCommands(commands=[])[source]

See: https://core.telegram.org/bots/api#setmycommands

getMyCommands()[source]

See: https://core.telegram.org/bots/api#getmycommands

setPassportDataErrors(user_id, errors)[source]

See: https://core.telegram.org/bots/api#setpassportdataerrors

answerPreCheckoutQuery(pre_checkout_query_id, ok, error_message=None)[source]

See: https://core.telegram.org/bots/api#answerprecheckoutquery

editMessageText(msg_identifier, text, parse_mode=None, disable_web_page_preview=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#editmessagetext

Parameters:msg_identifier – a 2-tuple (chat_id, message_id), a 1-tuple (inline_message_id), or simply inline_message_id. You may extract this value easily with amanobot.message_identifier()
editMessageCaption(msg_identifier, caption=None, parse_mode=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#editmessagecaption

Parameters:msg_identifier – Same as msg_identifier in amanobot.Bot.editMessageText()
editMessageMedia(msg_identifier, media, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#editmessagemedia

Parameters:msg_identifier – Same as msg_identifier in amanobot.Bot.editMessageText()
editMessageReplyMarkup(msg_identifier, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#editmessagereplymarkup

Parameters:msg_identifier – Same as msg_identifier in amanobot.Bot.editMessageText()
stopPoll(msg_identifier, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#stoppoll

Parameters:msg_identifier – a 2-tuple (chat_id, message_id). You may extract this value easily with amanobot.message_identifier()
deleteMessage(msg_identifier)[source]

See: https://core.telegram.org/bots/api#deletemessage

Parameters:msg_identifier – Same as msg_identifier in amanobot.Bot.editMessageText(), except this method does not work on inline messages.
sendSticker(chat_id, sticker, disable_notification=None, reply_to_message_id=None, reply_markup=None)[source]

See: https://core.telegram.org/bots/api#sendsticker

Parameters:sticker – Same as photo in amanobot.Bot.sendPhoto()
getStickerSet(name)[source]

See: https://core.telegram.org/bots/api#getstickerset

uploadStickerFile(user_id, png_sticker)[source]

See: https://core.telegram.org/bots/api#uploadstickerfile

createNewStickerSet(user_id, name, title, emojis, png_sticker=None, tgs_sticker=None, contains_masks=None, mask_position=None)[source]

See: https://core.telegram.org/bots/api#createnewstickerset

addStickerToSet(user_id, name, emojis, png_sticker=None, tgs_sticker=None, mask_position=None)[source]

See: https://core.telegram.org/bots/api#addstickertoset

setStickerPositionInSet(sticker, position)[source]

See: https://core.telegram.org/bots/api#setstickerpositioninset

deleteStickerFromSet(sticker)[source]

See: https://core.telegram.org/bots/api#deletestickerfromset

setStickerSetThumb(name, user_id, thumb=None)[source]

See: https://core.telegram.org/bots/api#setstickersetthumb

answerInlineQuery(inline_query_id, results, cache_time=None, is_personal=None, next_offset=None, switch_pm_text=None, switch_pm_parameter=None)[source]

See: https://core.telegram.org/bots/api#answerinlinequery

getUpdates(offset=None, limit=None, timeout=None, allowed_updates=None, _raise_errors=None)[source]

See: https://core.telegram.org/bots/api#getupdates

setWebhook(url=None, certificate=None, max_connections=None, allowed_updates=None)[source]

See: https://core.telegram.org/bots/api#setwebhook

deleteWebhook()[source]

See: https://core.telegram.org/bots/api#deletewebhook

getWebhookInfo()[source]

See: https://core.telegram.org/bots/api#getwebhookinfo

setGameScore(user_id, score, game_message_identifier, force=None, disable_edit_message=None)[source]

See: https://core.telegram.org/bots/api#setgamescore

Parameters:game_message_identifier – Same as msg_identifier in amanobot.Bot.editMessageText()
getGameHighScores(user_id, game_message_identifier)[source]

See: https://core.telegram.org/bots/api#getgamehighscores

Parameters:game_message_identifier – Same as msg_identifier in amanobot.Bot.editMessageText()
download_file(file_id, dest)[source]

Download a file to local disk.

Parameters:dest – a path or a file object
message_loop(callback=None, relax=0.1, timeout=20, allowed_updates=None, source=None, ordered=True, maxhold=3, run_forever=False)[source]
Deprecated:will be removed in future. Use MessageLoop instead.

Spawn a thread to constantly getUpdates or pull updates from a queue. Apply callback to every message received. Also starts the scheduler thread for internal events.

Parameters:callback – a function that takes one argument (the message), or a routing table. If None, the bot’s handle method is used.

A routing table is a dictionary of {flavor: function}, mapping messages to appropriate handler functions according to their flavors. It allows you to define functions specifically to handle one flavor of messages. It usually looks like this: {'chat': fn1, 'callback_query': fn2, 'inline_query': fn3, ...}. Each handler function should take one argument (the message).

Parameters:source – Source of updates. If None, getUpdates is used to obtain new messages from Telegram servers. If it is a synchronized queue (Queue.Queue in Python 2.7 or queue.Queue in Python 3), new messages are pulled from the queue. A web application implementing a webhook can dump updates into the queue, while the bot pulls from it. This is how amanobot can be integrated with webhooks.

Acceptable contents in queue:

  • str, unicode (Python 2.7), or bytes (Python 3, decoded using UTF-8) representing a JSON-serialized Update object.
  • a dict representing an Update object.

When source is None, these parameters are meaningful:

Parameters:
  • relax (float) – seconds between each getUpdates
  • timeout (int) – timeout parameter supplied to amanobot.Bot.getUpdates(), controlling how long to poll.
  • allowed_updates (array of string) – allowed_updates parameter supplied to amanobot.Bot.getUpdates(), controlling which types of updates to receive.

When source is a queue, these parameters are meaningful:

Parameters:
  • ordered (bool) – If True, ensure in-order delivery of messages to callback (i.e. updates with a smaller update_id always come before those with a larger update_id). If False, no re-ordering is done. callback is applied to messages as soon as they are pulled from queue.
  • maxhold (float) – Applied only when ordered is True. The maximum number of seconds an update is held waiting for a not-yet-arrived smaller update_id. When this number of seconds is up, the update is delivered to callback even if some smaller update_ids have not yet arrived. If those smaller update_ids arrive at some later time, they are discarded.

Finally, there is this parameter, meaningful always:

Parameters:run_forever (bool or str) – If True or any non-empty string, append an infinite loop at the end of this method, so it never returns. Useful as the very last line in a program. A non-empty string will also be printed, useful as an indication that the program is listening.

Message Loop and Webhook

There are two ways to obtain updates from Telegram Bot API: make calls to Bot.getUpdates() continuously, or use webhook.

In the former case, it is troublesome to have to program that manually. So MessageLoop is here to ease your burden. In the latter case, although the programming overhead is mainly on the web server, a structured way to funnel web requests into amanobot is desirable. The result is Webhook and OrderedWebhook.

The idea is similar. You supply a message-handling function to the object constructor, then use run_as_thread() to get it going. A MessageLoop makes calls to getUpdates() continuously, and apply the message-handling function to every message received. A Webhook or OrderedWebhook would not do anything by itself; you have to feed() it the new update every time the web server receives one.

In place of the message-handling function, you can supply one of the following:

  • a function that takes one argument (the message)
  • if None, the bot’s handle method is used
  • a routing table

A routing table is a dictionary of {flavor: function}, mapping messages to appropriate handler functions according to their flavors. It allows you to define functions specifically to handle one flavor of messages. It usually looks like this: {'chat': fn1, 'callback_query': fn2, 'inline_query': fn3, ...}. Each handler function should take one argument (the message).

class amanobot.loop.MessageLoop(bot, handle=None)[source]
run_forever(*args, **kwargs)[source]
Parameters:
  • relax (float) – seconds between each getUpdates()
  • offset (int) – initial offset parameter supplied to getUpdates()
  • timeout (int) – timeout parameter supplied to getUpdates(), controlling how long to poll.
  • allowed_updates (array of string) – allowed_updates parameter supplied to getUpdates(), controlling which types of updates to receive.

Calling this method will block forever. Use run_as_thread() to run it non-blockingly.

run_as_thread(*args, **kwargs)

In practice, you should always use OrderedWebhook rather than Webhook. Updates are individual HTTP requests, and there is no guarantee of their arrival order. OrderedWebhook puts them in order (according to update_id) before applying the message-handling function. In contrast, Webhook applies the message-handling function in the order you feed them. Unless you want to implement your own ordering logic, Webhook should not be used.

In async version, a task of run_forever() should be created instead of run_as_thread().

Refer to webhook examples for usage.

class amanobot.loop.OrderedWebhook(bot, handle=None)[source]
run_forever(*args, **kwargs)[source]
Parameters:maxhold (float) – The maximum number of seconds an update is held waiting for a not-yet-arrived smaller update_id. When this number of seconds is up, the update is delivered to the message-handling function even if some smaller update_ids have not yet arrived. If those smaller update_ids arrive at some later time, they are discarded.

Calling this method will block forever. Use run_as_thread() to run it non-blockingly.

feed(data)[source]
Parameters:data –

One of these:

  • str, unicode (Python 2.7), or bytes (Python 3, decoded using UTF-8) representing a JSON-serialized Update object.
  • a dict representing an Update object.
run_as_thread(*args, **kwargs)
class amanobot.loop.Webhook(bot, handle=None)[source]
run_forever()[source]
feed(data)[source]
run_as_thread(*args, **kwargs)

Functions

amanobot.flavor(msg)[source]

Return flavor of message or event.

A message’s flavor may be one of these:

  • chat
  • callback_query
  • inline_query
  • chosen_inline_result
  • shipping_query
  • pre_checkout_query

An event’s flavor is determined by the single top-level key.

amanobot.glance(msg, flavor='chat', long=False)[source]

Extract “headline” info about a message. Use parameter long to control whether a short or long tuple is returned.

When flavor is chat (msg being a Message object):

  • short: (content_type, msg['chat']['type'], msg['chat']['id'])
  • long: (content_type, msg['chat']['type'], msg['chat']['id'], msg['date'], msg['message_id'])

content_type can be: text, audio, document, game, photo, sticker, video, voice, video_note, contact, location, venue, new_chat_member, left_chat_member, new_chat_title, new_chat_photo, delete_chat_photo, group_chat_created, supergroup_chat_created, channel_chat_created, migrate_to_chat_id, migrate_from_chat_id, pinned_message, new_chat_members, invoice, successful_payment.

When flavor is callback_query (msg being a CallbackQuery object):

  • regardless: (msg['id'], msg['from']['id'], msg['data'])

When flavor is inline_query (msg being a InlineQuery object):

  • short: (msg['id'], msg['from']['id'], msg['query'])
  • long: (msg['id'], msg['from']['id'], msg['query'], msg['offset'])

When flavor is chosen_inline_result (msg being a ChosenInlineResult object):

  • regardless: (msg['result_id'], msg['from']['id'], msg['query'])

When flavor is shipping_query (msg being a ShippingQuery object):

  • regardless: (msg['id'], msg['from']['id'], msg['invoice_payload'])

When flavor is pre_checkout_query (msg being a PreCheckoutQuery object):

  • short: (msg['id'], msg['from']['id'], msg['invoice_payload'])
  • long: (msg['id'], msg['from']['id'], msg['invoice_payload'], msg['currency'], msg['total_amount'])
amanobot.flance(msg, long=False)[source]

A combination of amanobot.flavor() and amanobot.glance(), return a 2-tuple (flavor, headline_info), where headline_info is whatever extracted by amanobot.glance() depending on the message flavor and the long parameter.

amanobot.peel(event)[source]

Remove an event’s top-level skin (where its flavor is determined), and return the core content.

amanobot.fleece(event)[source]

A combination of amanobot.flavor() and amanobot.peel(), return a 2-tuple (flavor, content) of an event.

amanobot.is_event(msg)[source]

Return whether the message looks like an event. That is, whether it has a flavor that starts with an underscore.

amanobot.message_identifier(msg)[source]

Extract an identifier for message editing. Useful with amanobot.Bot.editMessageText() and similar methods. Returned value is guaranteed to be a tuple.

msg is expected to be chat or choson_inline_result.

amanobot.origin_identifier(msg)[source]

Extract the message identifier of a callback query’s origin. Returned value is guaranteed to be a tuple.

msg is expected to be callback_query.

DelegatorBot

class amanobot.DelegatorBot(token, delegation_patterns)[source]
Parameters:delegation_patterns – a list of (seeder, delegator) tuples.

A seeder is a function that:

  • takes one argument - a message

  • returns a seed. Depending on the nature of the seed, behavior is as follows:

    • if the seed is a hashable (e.g. number, string, tuple), it looks for a delegate associated with the seed. (Think of a dictionary of {seed: delegate})

      • if such a delegate exists and is alive, it is assumed that the message will be picked up by the delegate. Nothing more is done.
      • if no delegate exists or that delegate is no longer alive, a new delegate is obtained by calling the delegator function. The new delegate is associated with the seed.
      • In essence, when the seed is a hashable, only one delegate is running for a given seed.

    • if the seed is a non-hashable, (e.g. list), a new delegate is always obtained by calling the delegator function. No seed-delegate association occurs.

    • if the seed is None, nothing is done.

A delegator is a function that:

  • takes one argument - a (bot, message, seed) tuple. This is called a seed tuple.
  • returns a delegate, which can be one of the following:
    • an object that has methods start() and is_alive(). Therefore, a threading.Thread object is a natural delegate. Once returned, the object’s start() method is called.
    • a function. Once returned, it is wrapped in a Thread(target=function) and started.
    • a (function, args, kwargs) tuple. Once returned, it is wrapped in a Thread(target=function, args=args, kwargs=kwargs) and started.

The above logic is implemented in the handle method. You only have to create a MessageLoop with no callback argument, the above logic will be executed for every message received.

In the list of delegation patterns, all seeder functions are evaluated in order. One message may start multiple delegates.

The module amanobot.delegate has a bunch of seeder factories and delegator factories, which greatly ease the use of DelegatorBot. The module amanobot.helper also has a number of *Handler classes which provide a connection-like interface to deal with individual chats or users.

I have given an answer on Stack Overflow which elaborates on the inner workings of DelegatorBot in greater details. Interested readers are encouraged to read that.

In the rest of discussions, seed tuple means a (bot, message, seed) tuple, referring to the single argument taken by delegator functions.

amanobot.delegate

amanobot.delegate.per_chat_id(types='all')[source]
Parameters:types – all or a list of chat types (private, group, channel)
Returns:a seeder function that returns the chat id only if the chat type is in types.
amanobot.delegate.per_chat_id_in(s, types='all')[source]
Parameters:
  • s – a list or set of chat id
  • types – all or a list of chat types (private, group, channel)
Returns:

a seeder function that returns the chat id only if the chat id is in s and chat type is in types.
amanobot.delegate.per_chat_id_except(s, types='all')[source]
Parameters:
  • s – a list or set of chat id
  • types – all or a list of chat types (private, group, channel)
Returns:

a seeder function that returns the chat id only if the chat id is not in s and chat type is in types.
amanobot.delegate.per_from_id(flavors=['chat', 'inline_query', 'chosen_inline_result'])[source]
Parameters:flavors – all or a list of flavors
Returns:a seeder function that returns the from id only if the message flavor is in flavors.
amanobot.delegate.per_from_id_in(s, flavors=['chat', 'inline_query', 'chosen_inline_result'])[source]
Parameters:
  • s – a list or set of from id
  • flavors – all or a list of flavors
Returns:

a seeder function that returns the from id only if the from id is in s and message flavor is in flavors.
amanobot.delegate.per_from_id_except(s, flavors=['chat', 'inline_query', 'chosen_inline_result'])[source]
Parameters:
  • s – a list or set of from id
  • flavors – all or a list of flavors
Returns:

a seeder function that returns the from id only if the from id is not in s and message flavor is in flavors.
amanobot.delegate.per_inline_from_id()[source]
Returns:a seeder function that returns the from id only if the message flavor is inline_query or chosen_inline_result
amanobot.delegate.per_inline_from_id_in(s)[source]
Parameters:s – a list or set of from id
Returns:a seeder function that returns the from id only if the message flavor is inline_query or chosen_inline_result and the from id is in s.
amanobot.delegate.per_inline_from_id_except(s)[source]
Parameters:s – a list or set of from id
Returns:a seeder function that returns the from id only if the message flavor is inline_query or chosen_inline_result and the from id is not in s.
amanobot.delegate.per_application()[source]
Returns:a seeder function that always returns 1, ensuring at most one delegate is ever spawned for the entire application.
amanobot.delegate.per_message(flavors='all')[source]
Parameters:flavors – all or a list of flavors
Returns:a seeder function that returns a non-hashable only if the message flavor is in flavors.
amanobot.delegate.per_event_source_id(event_space)[source]
Returns:a seeder function that returns an event’s source id only if that event’s source space equals to event_space.
amanobot.delegate.per_callback_query_chat_id(types='all')[source]
Parameters:types – all or a list of chat types (private, group, channel)
Returns:a seeder function that returns a callback query’s originating chat id if the chat type is in types.
amanobot.delegate.per_callback_query_origin(origins='all')[source]
Parameters:origins – all or a list of origin types (chat, inline)
Returns:a seeder function that returns a callback query’s origin identifier if that origin type is in origins. The origin identifier is guaranteed to be a tuple.
amanobot.delegate.per_invoice_payload()[source]
Returns:a seeder function that returns the invoice payload.
amanobot.delegate.call(func, *args, **kwargs)[source]
Returns:a delegator function that returns a tuple (func, (seed tuple,)+ args, kwargs). That is, seed tuple is inserted before supplied positional arguments. By default, a thread wrapping func and all those arguments is spawned.
amanobot.delegate.create_run(cls, *args, **kwargs)[source]
Returns:a delegator function that calls the cls constructor whose arguments being a seed tuple followed by supplied *args and **kwargs, then returns the object’s run method. By default, a thread wrapping that run method is spawned.
amanobot.delegate.create_open(cls, *args, **kwargs)[source]
Returns:a delegator function that calls the cls constructor whose arguments being a seed tuple followed by supplied *args and **kwargs, then returns a looping function that uses the object’s listener to wait for messages and invokes instance method open, on_message, and on_close accordingly. By default, a thread wrapping that looping function is spawned.
amanobot.delegate.until(condition, fns)[source]

Try a list of seeder functions until a condition is met.

Parameters:
  • condition – a function that takes one argument - a seed - and returns True or False
  • fns – a list of seeder functions
Returns:

a “composite” seeder function that calls each supplied function in turn, and returns the first seed where the condition is met. If the condition is never met, it returns None.
amanobot.delegate.chain(*fns)[source]
Returns:a “composite” seeder function that calls each supplied function in turn, and returns the first seed that is not None.
amanobot.delegate.pair(seeders, delegator_factory, *args, **kwargs)[source]

The basic pair producer.

Returns:a (seeder, delegator_factory(*args, **kwargs)) tuple.
Parameters:seeders – If it is a seeder function or a list of one seeder function, it is returned as the final seeder. If it is a list of more than one seeder function, they are chained together before returned as the final seeder.
amanobot.delegate.pave_event_space(fn=<function pair>)[source]
Returns:a pair producer that ensures the seeder and delegator share the same event space.
amanobot.delegate.include_callback_query_chat_id(fn=<function pair>, types='all')[source]
Returns:a pair producer that enables static callback query capturing across seeder and delegator.
Parameters:types – all or a list of chat types (private, group, channel)
amanobot.delegate.intercept_callback_query_origin(fn=<function pair>, origins='all')[source]
Returns:a pair producer that enables dynamic callback query origin mapping across seeder and delegator.
Parameters:origins – all or a list of origin types (chat, inline). Origin mapping is only enabled for specified origin types.

amanobot.helper

Handlers

class amanobot.helper.Monitor(seed_tuple, capture, **kwargs)[source]

Bases: amanobot.helper.ListenerContext, amanobot.helper.DefaultRouterMixin

A delegate that never times-out, probably doing some kind of background monitoring in the application. Most naturally paired with per_application().

Parameters:capture – a list of patterns for Listener to capture
class amanobot.helper.ChatHandler(seed_tuple, include_callback_query=False, **kwargs)[source]

Bases: amanobot.helper.ChatContext, amanobot.helper.DefaultRouterMixin, amanobot.helper.StandardEventMixin, amanobot.helper.IdleTerminateMixin

A delegate to handle a chat.

class amanobot.helper.UserHandler(seed_tuple, include_callback_query=False, flavors=['chat', 'inline_query', 'chosen_inline_result'], **kwargs)[source]

Bases: amanobot.helper.UserContext, amanobot.helper.DefaultRouterMixin, amanobot.helper.StandardEventMixin, amanobot.helper.IdleTerminateMixin

A delegate to handle a user’s actions.

Parameters:flavors – A list of flavors to capture. all covers all flavors.
class amanobot.helper.InlineUserHandler(seed_tuple, **kwargs)[source]

Bases: amanobot.helper.UserHandler

A delegate to handle a user’s inline-related actions.

class amanobot.helper.CallbackQueryOriginHandler(seed_tuple, **kwargs)[source]

Bases: amanobot.helper.CallbackQueryOriginContext, amanobot.helper.DefaultRouterMixin, amanobot.helper.StandardEventMixin, amanobot.helper.IdleTerminateMixin

A delegate to handle callback query from one origin.

class amanobot.helper.InvoiceHandler(seed_tuple, **kwargs)[source]

Bases: amanobot.helper.InvoiceContext, amanobot.helper.DefaultRouterMixin, amanobot.helper.StandardEventMixin, amanobot.helper.IdleTerminateMixin

A delegate to handle messages related to an invoice.

Contexts

class amanobot.helper.ListenerContext(bot, context_id, *args, **kwargs)[source]
bot

The underlying Bot or an augmented version thereof

id
listener

See Listener

class amanobot.helper.ChatContext(bot, context_id, *args, **kwargs)[source]

Bases: amanobot.helper.ListenerContext

chat_id
sender

A Sender for this chat

administrator

An Administrator for this chat

class amanobot.helper.UserContext(bot, context_id, *args, **kwargs)[source]

Bases: amanobot.helper.ListenerContext

user_id
sender

A Sender for this user

class amanobot.helper.CallbackQueryOriginContext(bot, context_id, *args, **kwargs)[source]

Bases: amanobot.helper.ListenerContext

origin

Mesasge identifier of callback query’s origin

editor

An Editor to the originating message

class amanobot.helper.InvoiceContext(bot, context_id, *args, **kwargs)[source]

Bases: amanobot.helper.ListenerContext

payload
class amanobot.helper.Sender(bot, chat_id)[source]

When you are dealing with a particular chat, it is tedious to have to supply the same chat_id every time to send a message, or to send anything.

This object is a proxy to a bot’s send* and forwardMessage methods, automatically fills in a fixed chat id for you. Available methods have identical signatures as those of the underlying bot, except there is no need to supply the aforementioned chat_id:

class amanobot.helper.Administrator(bot, chat_id)[source]

When you are dealing with a particular chat, it is tedious to have to supply the same chat_id every time to get a chat’s info or to perform administrative tasks.

This object is a proxy to a bot’s chat administration methods, automatically fills in a fixed chat id for you. Available methods have identical signatures as those of the underlying bot, except there is no need to supply the aforementioned chat_id:

class amanobot.helper.Editor(bot, msg_identifier)[source]

If you want to edit a message over and over, it is tedious to have to supply the same msg_identifier every time.

This object is a proxy to a bot’s message-editing methods, automatically fills in a fixed message identifier for you. Available methods have identical signatures as those of the underlying bot, except there is no need to supply the aforementioned msg_identifier:

A message’s identifier can be easily extracted with amanobot.message_identifier().

Parameters:msg_identifier – a message identifier as mentioned above, or a message (whose identifier will be automatically extracted).
class amanobot.helper.Listener(mic, q)[source]
capture(pattern)[source]

Add a pattern to capture.

Parameters:pattern – a list of templates.
A template may be a function that:
  • takes one argument - a message
  • returns True to indicate a match
A template may also be a dictionary whose:
  • keys are used to select parts of message. Can be strings or regular expressions (as obtained by re.compile())
  • values are used to match against the selected parts. Can be typical data or a function.

All templates must produce a match for a message to be considered a match.

wait()[source]

Block until a matched message appears.

Mixins

class amanobot.helper.Router(key_function, routing_table)[source]

Map a message to a handler function, using a key function and a routing table (dictionary).

A key function digests a message down to a value. This value is treated as a key to the routing table to look up a corresponding handler function.

Parameters:
  • key_function –

    A function that takes one argument (the message) and returns one of the following:

    • a key to the routing table
    • a 1-tuple (key,)
    • a 2-tuple (key, (positional, arguments, …))
    • a 3-tuple (key, (positional, arguments, …), {keyword: arguments, …})

    Extra arguments, if returned, will be applied to the handler function after using the key to look up the routing table.

  • routing_table – A dictionary of {key: handler}. A None key acts as a default catch-all. If the key being looked up does not exist in the routing table, the None key and its corresponding handler is used.
map(msg)[source]

Apply key function to msg to obtain a key. Return the routing table entry.

route(msg, *aa, **kw)[source]

Apply key function to msg to obtain a key, look up routing table to obtain a handler function, then call the handler function with positional and keyword arguments, if any is returned by the key function.

*aa and **kw are dummy placeholders for easy chaining. Regardless of any number of arguments returned by the key function, multi-level routing may be achieved like this:

top_router.routing_table['key1'] = sub_router1.route
top_router.routing_table['key2'] = sub_router2.route
class amanobot.helper.DefaultRouterMixin(*args, **kwargs)[source]

Install a default Router and the instance method on_message().

router
on_message(msg)[source]

Call Router.route() to handle the message.

class amanobot.helper.StandardEventScheduler(scheduler, event_space, source_id)[source]

A proxy to the underlying Bot’s scheduler, this object implements the standard event format. A standard event looks like this:

{'_flavor': {
    'source': {
        'space': event_space, 'id': source_id}
    'custom_key1': custom_value1,
    'custom_key2': custom_value2,
     ... }}
  • There is a single top-level key indicating the flavor, starting with an _underscore.
  • On the second level, there is a source key indicating the event source.
  • An event source consists of an event space and a source id.
  • An event space is shared by all delegates in a group. Source id simply refers to a delegate’s id. They combine to ensure a delegate is always able to capture its own events, while its own events would not be mistakenly captured by others.

Events scheduled through this object always have the second-level source key fixed, while the flavor and other data may be customized.

event_space
configure(listener)[source]

Configure a Listener to capture events with this object’s event space and source id.

make_event_data(flavor, data)[source]

Marshall flavor and data into a standard event.

event_at(when, data_tuple)[source]

Schedule an event to be emitted at a certain time.

Parameters:
  • when – an absolute timestamp
  • data_tuple – a 2-tuple (flavor, data)
Returns:

an event object, useful for cancelling.
event_later(delay, data_tuple)[source]

Schedule an event to be emitted after a delay.

Parameters:
  • delay – number of seconds
  • data_tuple – a 2-tuple (flavor, data)
Returns:

an event object, useful for cancelling.
event_now(data_tuple)[source]

Schedule an event to be emitted now.

Parameters:data_tuple – a 2-tuple (flavor, data)
Returns:an event object, useful for cancelling.
cancel(event)[source]

Cancel an event.

class amanobot.helper.StandardEventMixin(event_space, *args, **kwargs)[source]

Install a StandardEventScheduler.

scheduler
class amanobot.helper.IdleEventCoordinator(scheduler, timeout)[source]
refresh()[source]

Refresh timeout timer

augment_on_message(handler)[source]
Returns:a function wrapping handler to refresh timer for every non-event message
augment_on_close(handler)[source]
Returns:a function wrapping handler to cancel timeout event
class amanobot.helper.IdleTerminateMixin(timeout, *args, **kwargs)[source]

Install an IdleEventCoordinator to manage idle timeout. Also define instance method on__idle() to handle idle timeout events.

idle_event_coordinator
on__idle(event)[source]

Raise an IdleTerminate to close the delegate.

class amanobot.helper.CallbackQueryCoordinator(id, origin_set, enable_chat, enable_inline)[source]
Parameters:
  • origin_set – Callback query whose origin belongs to this set will be captured
  • enable_chat –
    • False: Do not intercept chat-originated callback query
    • True: Do intercept
    • Notifier function: Do intercept and call the notifier function on adding or removing an origin
  • enable_inline – Same meaning as enable_chat, but apply to inline-originated callback query

Notifier functions should have the signature notifier(origin, id, adding):

  • On adding an origin, notifier(origin, my_id, True) will be called.
  • On removing an origin, notifier(origin, my_id, False) will be called.
configure(listener)[source]

Configure a Listener to capture callback query

capture_origin(msg_identifier, notify=True)[source]
uncapture_origin(msg_identifier, notify=True)[source]
augment_send(send_func)[source]
Parameters:send_func – a function that sends messages, such as Bot.send*()
Returns:a function that wraps around send_func and examines whether the sent message contains an inline keyboard with callback data. If so, future callback query originating from the sent message will be captured.
augment_edit(edit_func)[source]
Parameters:edit_func – a function that edits messages, such as Bot.edit*()
Returns:a function that wraps around edit_func and examines whether the edited message contains an inline keyboard with callback data. If so, future callback query originating from the edited message will be captured. If not, such capturing will be stopped.
augment_delete(delete_func)[source]
Parameters:delete_func – a function that deletes messages, such as Bot.deleteMessage()
Returns:a function that wraps around delete_func and stops capturing callback query originating from that deleted message.
augment_on_message(handler)[source]
Parameters:handler – an on_message() handler function
Returns:a function that wraps around handler and examines whether the incoming message is a chosen inline result with an inline_message_id field. If so, future callback query originating from this chosen inline result will be captured.
augment_bot(bot)[source]
Returns:a proxy to bot with these modifications:
class amanobot.helper.InterceptCallbackQueryMixin(intercept_callback_query, *args, **kwargs)[source]

Install a CallbackQueryCoordinator to capture callback query dynamically.

Using this mixin has one consequence. The self.bot() property no longer returns the original Bot object. Instead, it returns an augmented version of the Bot (augmented by CallbackQueryCoordinator). The original Bot can be accessed with self.__bot (double underscore).

Parameters:intercept_callback_query – a 2-tuple (enable_chat, enable_inline) to pass to CallbackQueryCoordinator
callback_query_coordinator
class amanobot.helper.Answerer(bot)[source]

When processing inline queries, ensure at most one active thread per user id.

answer(inline_query, compute_fn, *compute_args, **compute_kwargs)[source]

Spawns a thread that calls compute fn (along with additional arguments *compute_args and **compute_kwargs), then applies the returned value to Bot.answerInlineQuery() to answer the inline query. If a preceding thread is already working for a user, that thread is cancelled, thus ensuring at most one active thread per user id.

Parameters:
  • inline_query – The inline query to be processed. The originating user is inferred from msg['from']['id'].
  • compute_fn –

    A thread-safe function whose returned value is given to Bot.answerInlineQuery() to send. May return:

  • *compute_args – positional arguments to compute_fn
  • **compute_kwargs – keyword arguments to compute_fn
class amanobot.helper.AnswererMixin(*args, **kwargs)[source]

Install an Answerer to handle inline query.

answerer

Utilities

class amanobot.helper.SafeDict(*args, **kwargs)[source]

A subclass of dict, thread-safety added:

d = SafeDict()  # Thread-safe operations include:
d['a'] = 3      # key assignment
d['a']          # key retrieval
del d['a']      # key deletion
amanobot.helper.openable(cls)[source]

A class decorator to fill in certain methods and properties to ensure a class can be used by create_open().

These instance methods and property will be added, if not defined by the class:

  • open(self, initial_msg, seed)
  • on_message(self, msg)
  • on_close(self, ex)
  • close(self, ex=None)
  • property listener

amanobot.exception

exception amanobot.exception.AmanobotException[source]

Base class of following exceptions.

exception amanobot.exception.BadFlavor(offender)[source]
offender
exception amanobot.exception.BadHTTPResponse(status, text, response)[source]

All requests to Bot API should result in a JSON response. If non-JSON, this exception is raised. While it is hard to pinpoint exactly when this might happen, the following situations have been observed to give rise to it:

  • an unreasonable token, e.g. abc, 123, anything that does not even remotely resemble a correct token.
  • a bad gateway, e.g. when Telegram servers are down.
status
text
response
exception amanobot.exception.EventNotFound(event)[source]
event
exception amanobot.exception.WaitTooLong(seconds)[source]
seconds
exception amanobot.exception.IdleTerminate(seconds)[source]
exception amanobot.exception.StopListening[source]
exception amanobot.exception.TelegramError(description, error_code, json)[source]

To indicate erroneous situations, Telegram returns a JSON object containing an error code and a description. This will cause a TelegramError to be raised. Before raising a generic TelegramError, amanobot looks for a more specific subclass that “matches” the error. If such a class exists, an exception of that specific subclass is raised. This allows you to either catch specific errors or to cast a wide net (by a catch-all TelegramError). This also allows you to incorporate custom TelegramError easily.

Subclasses must define a class variable DESCRIPTION_PATTERNS which is a list of regular expressions. If an error’s description matches any of the regular expressions, an exception of that subclass is raised.

description
error_code
json
exception amanobot.exception.UnauthorizedError(description, error_code, json)[source]
DESCRIPTION_PATTERNS = ['unauthorized']
exception amanobot.exception.BotWasKickedError(description, error_code, json)[source]
DESCRIPTION_PATTERNS = ['bot.*kicked']
exception amanobot.exception.BotWasBlockedError(description, error_code, json)[source]
DESCRIPTION_PATTERNS = ['bot.*blocked']
exception amanobot.exception.TooManyRequestsError(description, error_code, json)[source]
DESCRIPTION_PATTERNS = ['too *many *requests']
exception amanobot.exception.MigratedToSupergroupChatError(description, error_code, json)[source]
DESCRIPTION_PATTERNS = ['migrated.*supergroup *chat']
exception amanobot.exception.NotEnoughRightsError(description, error_code, json)[source]
DESCRIPTION_PATTERNS = ['not *enough *rights']

amanobot.namedtuple

Amanobot’s custom is to represent Bot API object as dictionary. On the other hand, the module amanobot.namedtuple also provide namedtuple classes mirroring those objects. The reasons are twofold:

  1. Under some situations, you may want an object with a complete set of fields, including those whose values are None. A dictionary translated from Bot API’s response would have those None fields absent. By converting such a dictionary to a namedtuple, all fields are guaranteed to be present, even if their values are None. This usage is for incoming objects received from Telegram servers.
  2. Namedtuple allows easier construction of objects like ReplyKeyboardMarkup, InlineKeyboardMarkup, and various InlineQueryResult, etc. This usage is for outgoing objects sent to Telegram servers.

Incoming objects include:

Outgoing objects include:

amanobot.routing

This module has a bunch of key function factories and routing table factories to facilitate the use of Router.

Things to remember:

  1. A key function takes one argument - the message, and returns a key, optionally followed by positional arguments and keyword arguments.
  2. A routing table is just a dictionary. After obtaining one from a factory function, you can customize it to your liking.
amanobot.routing.by_content_type()[source]
Returns:A key function that returns a 2-tuple (content_type, (msg[content_type],)). In plain English, it returns the message’s content type as the key, and the corresponding content as a positional argument to the handler function.
amanobot.routing.by_command(extractor, prefix=('/', ), separator=' ', pass_args=False)[source]
Parameters:
  • extractor – a function that takes one argument (the message) and returns a portion of message to be interpreted. To extract the text of a chat message, use lambda msg: msg['text'].
  • prefix – a list of special characters expected to indicate the head of a command.
  • separator – a command may be followed by arguments separated by separator.
  • pass_args (bool) – If True, arguments following a command will be passed to the handler function.
Returns:

a key function that interprets a specific part of a message and returns the embedded command, optionally followed by arguments. If the text is not preceded by any of the specified prefix, it returns a 1-tuple (None,) as the key. This is to distinguish with the special None key in routing table.
amanobot.routing.by_chat_command(prefix=('/', ), separator=' ', pass_args=False)[source]
Parameters:
  • prefix – a list of special characters expected to indicate the head of a command.
  • separator – a command may be followed by arguments separated by separator.
  • pass_args (bool) – If True, arguments following a command will be passed to the handler function.
Returns:

a key function that interprets a chat message’s text and returns the embedded command, optionally followed by arguments. If the text is not preceded by any of the specified prefix, it returns a 1-tuple (None,) as the key. This is to distinguish with the special None key in routing table.
amanobot.routing.by_text()[source]
Returns:a key function that returns a message’s text field.
amanobot.routing.by_data()[source]
Returns:a key function that returns a message’s data field.
amanobot.routing.by_regex(extractor, regex, key=1)[source]
Parameters:
  • extractor – a function that takes one argument (the message) and returns a portion of message to be interpreted. To extract the text of a chat message, use lambda msg: msg['text'].
  • regex (str or regex object) – the pattern to look for
  • key – the part of match object to be used as key
Returns:

a key function that returns match.group(key) as key (where match is the match object) and the match object as a positional argument. If no match is found, it returns a 1-tuple (None,) as the key. This is to distinguish with the special None key in routing table.
amanobot.routing.process_key(processor, fn)[source]
Parameters:
  • processor – a function to process the key returned by the supplied key function
  • fn – a key function
Returns:

a function that wraps around the supplied key function to further process the key before returning.
amanobot.routing.lower_key(fn)[source]
Parameters:fn – a key function
Returns:a function that wraps around the supplied key function to ensure the returned key is in lowercase.
amanobot.routing.upper_key(fn)[source]
Parameters:fn – a key function
Returns:a function that wraps around the supplied key function to ensure the returned key is in uppercase.
amanobot.routing.make_routing_table(obj, keys, prefix='on_')[source]
Returns:

a dictionary roughly equivalent to {'key1': obj.on_key1, 'key2': obj.on_key2, ...}, but obj does not have to define all methods. It may define the needed ones only.
Parameters:
  • obj – the object
  • keys – a list of keys
  • prefix – a string to be prepended to keys to make method names
amanobot.routing.make_content_type_routing_table(obj, prefix='on_')[source]
Returns:

a dictionary covering all available content types, roughly equivalent to {'text': obj.on_text, 'photo': obj.on_photo, ...}, but obj does not have to define all methods. It may define the needed ones only.
Parameters:
  • obj – the object
  • prefix – a string to be prepended to content types to make method names

amanobot.text

amanobot.text.apply_entities_as_markdown(text, entities)[source]

Format text as Markdown. Also take care of escaping special characters. Returned value can be passed to Bot.sendMessage() with appropriate parse_mode.

Parameters:
  • text – plain text
  • entities –

    a list of MessageEntity objects
amanobot.text.apply_entities_as_html(text, entities)[source]

Format text as HTML. Also take care of escaping special characters. Returned value can be passed to Bot.sendMessage() with appropriate parse_mode.

Parameters:
  • text – plain text
  • entities –

    a list of MessageEntity objects

amanobot.api

amanobot.api.set_proxy(url, basic_auth=None)[source]

Access Bot API through a proxy.

Parameters:
  • url – proxy URL
  • basic_auth – 2-tuple ('username', 'password')

Amanobot helps you build applications for Telegram Bot API. It works on Python 2.7 and Python 3. For Python 3.5+, it also has an async version based on asyncio.

For a time, I tried to list the features here like many projects do. Eventually, I gave up.

Common and straight-forward features are too trivial to worth listing. For more unique and novel features, I cannot find standard terms to describe them. The best way to experience amanobot is by reading this page and going through the examples. Let’s go.

Installation

pip:

$ pip install amanobot
$ pip install amanobot --upgrade  # UPGRADE

easy_install:

$ easy_install amanobot
$ easy_install --upgrade amanobot  # UPGRADE

Get a token

To use the Telegram Bot API, you first have to get a bot account by chatting with BotFather.

BotFather will give you a token, something like 123456789:ABCdefGhIJKlmNoPQRsTUVwxyZ. With the token in hand, you can start using amanobot to access the bot account.

Test the account

>>> import amanobot
>>> bot = amanobot.Bot('PUT YOUR TOKEN HERE')
>>> bot.getMe()
{'id': 123456789, 'is_bot': True, 'first_name': 'Your Bot', 'username': 'YourBot'}

Receive messages

Bots cannot initiate conversations with users. You have to send it a message first. Get the message by calling Bot.getUpdates():

>>> from pprint import pprint
>>> response = bot.getUpdates()
>>> pprint(response)
[{'message': {'chat': {'first_name': '• Alisson™ •',
                   'id': 200097591,
                   'last_name': '🇧🇷',
                   'type': 'private',
                   'username': 'marminino'},
          'date': 1538346501,
          'from': {'first_name': '• Alisson™ •',
                   'id': 200097591,
                   'is_bot': False,
                   'language_code': 'pt-br',
                   'last_name': '🇧🇷',
                   'username': 'marminino'},
          'message_id': 149,
          'text': 'Hello'},
'update_id': 100000000}]

The chat field represents the conversation. Its type can be private, group, supergroup, or channel (whose meanings should be obvious, I hope). Above, • Alisson™ • just sent a private message to the bot.

According to Bot API, the method getUpdates returns an array of Update objects. As you can see, an Update object is nothing more than a Python dictionary. In amanobot, Bot API objects are represented as dictionary.

Note the update_id. It is an ever-increasing number. Next time you should use getUpdates(offset=100000001) to avoid getting the same old messages over and over. Giving an offset essentially acknowledges to the server that you have received all update_ids lower than offset:

>>> bot.getUpdates(offset=100000001)
[]

An easier way to receive messages

It is troublesome to keep checking messages while managing offset. Let amanobot take care of the mundane stuff and notify you whenever new messages arrive:

>>> from amanobot.loop import MessageLoop
>>> def handle(msg):
...     pprint(msg)
...
>>> MessageLoop(bot, handle).run_as_thread()

After setting this up, send it a few messages. Sit back and monitor the messages arriving.

Send a message

Sooner or later, your bot will want to send you messages. You should have discovered your own user id from above interactions. I will use my real id on this example. Remember to substitute your own id:

>>> bot.sendMessage(200097591, 'Hey!')

Quickly glance a message

When processing a message, a few pieces of information are so central that you almost always have to extract them. Use amanobot.glance() to extract “headline info”. Try this skeleton, a bot which echoes what you said:

import sys
import time
import amanobot
from amanobot.loop import MessageLoop

def handle(msg):
    content_type, chat_type, chat_id = amanobot.glance(msg)
    print(content_type, chat_type, chat_id)

    if content_type == 'text':
        bot.sendMessage(chat_id, msg['text'])

TOKEN = sys.argv[1]  # get token from command-line

bot = amanobot.Bot(TOKEN)
MessageLoop(bot, handle).run_as_thread()
print ('Listening ...')

# Keep the program running.
while 1:
    time.sleep(10)

It is a good habit to always check content_type before further processing. Do not assume every message is a text.

Custom Keyboard and Inline Keyboard

Besides sending messages back and forth, Bot API allows richer interactions with custom keyboard and inline keyboard. Both can be specified with the parameter reply_markup in Bot.sendMessage(). The module amanobot.namedtuple provides namedtuple classes for easier construction of these keyboards.

Pressing a button on a custom keyboard results in a Message object sent to the bot, which is no different from a regular chat message composed by typing.

Pressing a button on an inline keyboard results in a CallbackQuery object sent to the bot, which we have to distinguish from a Message object.

Here comes the concept of flavor.

Message has a Flavor

Regardless of the type of objects received, amanobot generically calls them “message” (with a lowercase “m”). A message’s flavor depends on the underlying object:

  • a Message object gives the flavor chat
  • a CallbackQuery object gives the flavor callback_query
  • there are two more flavors, which you will come to shortly.

Use amanobot.flavor() to check a message’s flavor.

Here is a bot which does two things:

  • When you send it a message, it gives you an inline keyboard.
  • When you press a button on the inline keyboard, it says “Got it”.

Pay attention to these things in the code:

import sys
import time
import amanobot
from amanobot.loop import MessageLoop
from amanobot.namedtuple import InlineKeyboardMarkup, InlineKeyboardButton

def on_chat_message(msg):
    content_type, chat_type, chat_id = amanobot.glance(msg)

    keyboard = InlineKeyboardMarkup(inline_keyboard=[
                   [InlineKeyboardButton(text='Press me', callback_data='press')],
               ])

    bot.sendMessage(chat_id, 'Use inline keyboard', reply_markup=keyboard)

def on_callback_query(msg):
    query_id, from_id, query_data = amanobot.glance(msg, flavor='callback_query')
    print('Callback Query:', query_id, from_id, query_data)

    bot.answerCallbackQuery(query_id, text='Got it')

TOKEN = sys.argv[1]  # get token from command-line

bot = amanobot.Bot(TOKEN)
MessageLoop(bot, {'chat': on_chat_message,
                  'callback_query': on_callback_query}).run_as_thread()
print('Listening ...')

while 1:
    time.sleep(10)

Inline Query

So far, the bot has been operating in a chat - private, group, or channel.

In a private chat, Alice talks to Bot. Simple enough.

In a group chat, Alice, Bot, and Charlie share the same group. As the humans gossip in the group, Bot hears selected messages (depending on whether in privacy mode or not) and may chime in once in a while.

Inline query is a totally different mode of operations.

Imagine this. Alice wants to recommend a restaurant to Zach, but she can’t remember the location right off her head. Inside the chat screen with Zach, Alice types @Bot where is my favorite restaurant, issuing an inline query to Bot, like asking Bot a question. Bot gives back a list of answers; Alice can choose one of them - as she taps on an answer, that answer is sent to Zach as a chat message. In this case, Bot never takes part in the conversation. Instead, Bot acts as an assistant, ready to give you talking materials. For every answer Alice chooses, Bot gets notified with a chosen inline result.

To enable a bot to receive InlineQuery, you have to send a /setinline command to BotFather. An InlineQuery message gives the flavor inline_query.

To enable a bot to receive ChosenInlineResult, you have to send a /setinlinefeedback command to BotFather. A ChosenInlineResult message gives the flavor chosen_inline_result.

In this code sample, pay attention to these things:

import sys
import time
import amanobot
from amanobot.loop import MessageLoop
from amanobot.namedtuple import InlineQueryResultArticle, InputTextMessageContent

def on_inline_query(msg):
    query_id, from_id, query_string = amanobot.glance(msg, flavor='inline_query')
    print ('Inline Query:', query_id, from_id, query_string)

    articles = [InlineQueryResultArticle(
                    id='abc',
                    title='ABC',
                    input_message_content=InputTextMessageContent(
                        message_text='Hello'
                    )
               )]

    bot.answerInlineQuery(query_id, articles)

def on_chosen_inline_result(msg):
    result_id, from_id, query_string = amanobot.glance(msg, flavor='chosen_inline_result')
    print ('Chosen Inline Result:', result_id, from_id, query_string)

TOKEN = sys.argv[1]  # get token from command-line

bot = amanobot.Bot(TOKEN)
MessageLoop(bot, {'inline_query': on_inline_query,
                  'chosen_inline_result': on_chosen_inline_result}).run_as_thread()

while 1:
    time.sleep(10)

However, this has a small problem. As you types and pauses, types and pauses, types and pauses … closely bunched inline queries arrive. In fact, a new inline query often arrives before we finish processing a preceding one. With only a single thread of execution, we can only process the closely bunched inline queries sequentially. Ideally, whenever we see a new inline query coming from the same user, it should override and cancel any preceding inline queries being processed (that belong to the same user).

My solution is this. An Answerer takes an inline query, inspects its from id (the originating user id), and checks to see whether that user has an unfinished thread processing a preceding inline query. If there is, the unfinished thread will be cancelled before a new thread is spawned to process the latest inline query. In other words, an Answerer ensures at most one active inline-query-processing thread per user.

Answerer also frees you from having to call Bot.answerInlineQuery() every time. You supply it with a compute function. It takes that function’s returned value and calls Bot.answerInlineQuery() to send the results. Being accessible by multiple threads, the compute function must be thread-safe.

import sys
import time
import amanobot
from amanobot.loop import MessageLoop
from amanobot.namedtuple import InlineQueryResultArticle, InputTextMessageContent

def on_inline_query(msg):
    def compute():
        query_id, from_id, query_string = amanobot.glance(msg, flavor='inline_query')
        print('Inline Query:', query_id, from_id, query_string)

        articles = [InlineQueryResultArticle(
                        id='abc',
                        title=query_string,
                        input_message_content=InputTextMessageContent(
                            message_text=query_string
                        )
                   )]

        return articles

    answerer.answer(msg, compute)

def on_chosen_inline_result(msg):
    result_id, from_id, query_string = amanobot.glance(msg, flavor='chosen_inline_result')
    print ('Chosen Inline Result:', result_id, from_id, query_string)

TOKEN = sys.argv[1]  # get token from command-line

bot = amanobot.Bot(TOKEN)
answerer = amanobot.helper.Answerer(bot)

MessageLoop(bot, {'inline_query': on_inline_query,
                  'chosen_inline_result': on_chosen_inline_result}).run_as_thread()

while 1:
    time.sleep(10)

Maintain Threads of Conversation

So far, we have been using a single line of execution to handle messages. That is adequate for simple programs. For more sophisticated programs where states need to be maintained across messages, a better approach is needed.

Consider this scenario. A bot wants to have an intelligent conversation with a lot of users, and if we could only use a single line of execution to handle messages (like what we have done so far), we would have to maintain some state variables about each conversation outside the message-handling function(s). On receiving each message, we first have to check whether the user already has a conversation started, and if so, what we have been talking about. To avoid such mundaneness, we need a structured way to maintain “threads” of conversation.

Let’s look at my solution. Here, I implemented a bot that counts how many messages have been sent by an individual user. If no message is received after 10 seconds, it starts over (timeout). The counting is done per chat - that’s the important point.

import sys
import time
import amanobot
from amanobot.loop import MessageLoop
from amanobot.delegate import pave_event_space, per_chat_id, create_open

class MessageCounter(amanobot.helper.ChatHandler):
    def __init__(self, *args, **kwargs):
        super(MessageCounter, self).__init__(*args, **kwargs)
        self._count = 0

    def on_chat_message(self, msg):
        self._count += 1
        self.sender.sendMessage(self._count)

TOKEN = sys.argv[1]  # get token from command-line

bot = amanobot.DelegatorBot(TOKEN, [
    pave_event_space()(
        per_chat_id(), create_open, MessageCounter, timeout=10),
])
MessageLoop(bot).run_as_thread()

while 1:
    time.sleep(10)

A DelegatorBot is able to spawn delegates. Above, it is spawning one MessageCounter per chat id.

Also noteworthy is pave_event_space(). To kill itself after 10 seconds of inactivity, the delegate schedules a timeout event. For events to work, we need to prepare an event space.

Detailed explanation of the delegation mechanism (e.g. how and when a MessageCounter is created, and why) is beyond the scope here. Please refer to DelegatorBot.

Inline Handler per User

You may also want to answer inline query differently depending on user. When Alice asks Bot “Where is my favorite restaurant?”, Bot should give a different answer than when Charlie asks the same question.

In the code sample below, pay attention to these things:

import sys
import time
import amanobot
from amanobot.loop import MessageLoop
from amanobot.delegate import pave_event_space, per_inline_from_id, create_open
from amanobot.namedtuple import InlineQueryResultArticle, InputTextMessageContent

class QueryCounter(amanobot.helper.InlineUserHandler, amanobot.helper.AnswererMixin):
    def __init__(self, *args, **kwargs):
        super(QueryCounter, self).__init__(*args, **kwargs)
        self._count = 0

    def on_inline_query(self, msg):
        def compute():
            query_id, from_id, query_string = amanobot.glance(msg, flavor='inline_query')
            print(self.id, ':', 'Inline Query:', query_id, from_id, query_string)

            self._count += 1
            text = '%d. %s' % (self._count, query_string)

            articles = [InlineQueryResultArticle(
                            id='abc',
                            title=text,
                            input_message_content=InputTextMessageContent(
                                message_text=text
                            )
                       )]

            return articles

        self.answerer.answer(msg, compute)

    def on_chosen_inline_result(self, msg):
        result_id, from_id, query_string = amanobot.glance(msg, flavor='chosen_inline_result')
        print(self.id, ':', 'Chosen Inline Result:', result_id, from_id, query_string)

TOKEN = sys.argv[1]  # get token from command-line

bot = amanobot.DelegatorBot(TOKEN, [
    pave_event_space()(
        per_inline_from_id(), create_open, QueryCounter, timeout=10),
])
MessageLoop(bot).run_as_thread()

while 1:
    time.sleep(10)

Async Version (Python 3.5+)

Everything discussed so far assumes traditional Python. That is, network operations are blocking; if you want to serve many users at the same time, some kind of threads are usually needed. Another option is to use an asynchronous or event-driven framework, such as Twisted.

Python 3.5+ has its own asyncio module. Amanobot supports that, too.

Here is how to compile and install Python 3.6, if your O/S does not have it built in:

$ sudo apt-get update
$ sudo apt-get upgrade
$ sudo apt-get install libssl-dev openssl libreadline-dev
$ cd ~
$ wget https://www.python.org/ftp/python/3.6.6/Python-3.6.6.tgz
$ tar zxf Python-3.6.6.tgz
$ cd Python-3.6.6
$ ./configure
$ make
$ sudo make install

Finally:

$ pip3.6 install amanobot

In case you are not familiar with asynchronous programming, let’s start by learning about generators and coroutines:

… why we want asynchronous programming:

… how generators and coroutines are applied to asynchronous programming:

… and how an asyncio program is generally structured:

Amanobot’s async version basically mirrors the traditional version. Main differences are:

  • blocking methods are now coroutines, and should be called with await
  • delegation is achieved by tasks, instead of threads

Because of that (and this is true of asynchronous Python in general), a lot of methods will not work in the interactive Python interpreter like regular functions would. They will have to be driven by an event loop.

Async version is under module amanobot.aio. I duplicate the message counter example below in async style:

import sys
import asyncio
import amanobot
from amanobot.aio.loop import MessageLoop
from amanobot.aio.delegate import pave_event_space, per_chat_id, create_open

class MessageCounter(amanobot.aio.helper.ChatHandler):
    def __init__(self, *args, **kwargs):
        super(MessageCounter, self).__init__(*args, **kwargs)
        self._count = 0

    async def on_chat_message(self, msg):
        self._count += 1
        await self.sender.sendMessage(self._count)

TOKEN = sys.argv[1]  # get token from command-line

bot = amanobot.aio.DelegatorBot(TOKEN, [
    pave_event_space()(
        per_chat_id(), create_open, MessageCounter, timeout=10),
])

loop = asyncio.get_event_loop()
loop.create_task(MessageLoop(bot).run_forever())
print('Listening ...')

loop.run_forever()