python/aiogram
1
0
Fork 0
mirror of https://github.com/aiogram/aiogram.git synced 2026-04-08 16:37:47 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

Merge remote-tracking branch 'origin/dev-3.x' into dev-3.x

Browse source 
# Conflicts:
#	README.md
#	README.rst
#	aiogram/__init__.py
#	aiogram/bot/bot.py
#	aiogram/contrib/fsm_storage/redis.py
#	aiogram/contrib/middlewares/logging.py
#	aiogram/dispatcher/dispatcher.py
#	aiogram/dispatcher/filters/__init__.py
#	aiogram/dispatcher/filters/builtin.py
#	aiogram/dispatcher/filters/filters.py
#	aiogram/dispatcher/filters/state.py
#	aiogram/dispatcher/handler.py
#	aiogram/dispatcher/webhook.py
#	aiogram/types/base.py
#	aiogram/types/chat.py
#	aiogram/types/chat_member.py
#	aiogram/types/input_media.py
#	aiogram/types/message.py
#	aiogram/utils/callback_data.py
#	aiogram/utils/deprecated.py
#	aiogram/utils/exceptions.py
#	aiogram/utils/executor.py
#	aiogram/utils/helper.py
#	aiogram/utils/json.py
#	aiogram/utils/mixins.py
#	aiogram/utils/payload.py
#	dev_requirements.txt
#	docs/source/index.rst
#	examples/callback_data_factory.py
#	examples/check_user_language.py
#	examples/echo_bot.py
#	examples/finite_state_machine_example.py
#	examples/i18n_example.py
#	examples/inline_bot.py
#	examples/media_group.py
#	examples/middleware_and_antiflood.py
#	examples/payments.py
#	examples/proxy_and_emojize.py
#	examples/regexp_commands_filter_example.py
#	examples/throtling_example.py
#	examples/webhook_example.py
#	examples/webhook_example_2.py
#	setup.py
#	tests/test_bot.py
#	tests/test_token.py
#	tests/types/dataset.py
This commit is contained in:
Alex Root Junior 2019-11-03 22:19:44 +02:00
commit 87393f2475
98 changed files with 5048 additions and 4854 deletions
Download patch file Download diff file Expand all files Collapse all files

33
examples/admin_filter_example.py Normal file
View file

@ -0,0 +1,33 @@
import logging
from aiogram import Bot, Dispatcher, types, executor
API_TOKEN = 'API_TOKEN_HERE'
logging.basicConfig(level=logging.DEBUG)
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot=bot)
# checks specified chat
@dp.message_handler(is_chat_admin=-1001241113577)
async def handle_specified(msg: types.Message):
await msg.answer("You are an admin of the specified chat!")
# checks multiple chats
@dp.message_handler(is_chat_admin=[-1001241113577, -320463906])
async def handle_multiple(msg: types.Message):
await msg.answer("You are an admin of multiple chats!")
# checks current chat
@dp.message_handler(is_chat_admin=True)
async def handler3(msg: types.Message):
await msg.answer("You are an admin of the current chat!")
if __name__ == '__main__':
executor.start_polling(dp)

4
examples/adwanced_executor_example.py → examples/advanced_executor_example.py
View file

@ -4,7 +4,7 @@
In this example used ArgumentParser for configuring Your bot.
Provided to start bot with webhook:
python adwanced_executor_example.py \
python advanced_executor_example.py \
--token TOKEN_HERE \
--host 0.0.0.0 \
--port 8084 \
@ -12,7 +12,7 @@ Provided to start bot with webhook:
--webhook-port 443
Or long polling:
python adwanced_executor_example.py --token TOKEN_HERE
python advanced_executor_example.py --token TOKEN_HERE
So... In this example found small trouble:
can't get bot instance in handlers.

5
examples/broadcast_example.py
View file

@ -9,9 +9,8 @@ API_TOKEN = "BOT TOKEN HERE"
logging.basicConfig(level=logging.INFO)
log = logging.getLogger("broadcast")
loop = asyncio.get_event_loop()
bot = Bot(token=API_TOKEN, loop=loop, parse_mode=types.ParseMode.HTML)
dp = Dispatcher(bot, loop=loop)
bot = Bot(token=API_TOKEN, parse_mode=types.ParseMode.HTML)
dp = Dispatcher(bot)
def get_users():

87
examples/callback_data_factory.py
View file

@ -1,4 +1,3 @@
import asyncio
import logging
import random
import uuid
@ -11,27 +10,26 @@ from aiogram.utils.exceptions import MessageNotModified, Throttled
logging.basicConfig(level=logging.INFO)
API_TOKEN = "BOT TOKEN HERE"
API_TOKEN = 'BOT TOKEN HERE'
loop = asyncio.get_event_loop()
bot = Bot(token=API_TOKEN, loop=loop, parse_mode=types.ParseMode.HTML)
bot = Bot(token=API_TOKEN, parse_mode=types.ParseMode.HTML)
storage = MemoryStorage()
dp = Dispatcher(bot, storage=storage)
dp.middleware.setup(LoggingMiddleware())
POSTS = {
str(uuid.uuid4()): {
"title": f"Post {index}",
"body": "Lorem ipsum dolor sit amet, "
"consectetur adipiscing elit, "
"sed do eiusmod tempor incididunt ut "
"labore et dolore magna aliqua",
"votes": random.randint(-2, 5),
}
for index in range(1, 6)
'title': f'Post {index}',
'body': 'Lorem ipsum dolor sit amet, '
'consectetur adipiscing elit, '
'sed do eiusmod tempor incididunt ut '
'labore et dolore magna aliqua',
'votes': random.randint(-2, 5),
} for index in range(1, 6)
}
posts_cb = CallbackData("post", "id", "action") # post:<id>:<action>
posts_cb = CallbackData('post', 'id', 'action') # post:<id>:<action>
def get_keyboard() -> types.InlineKeyboardMarkup:
@ -42,73 +40,72 @@ def get_keyboard() -> types.InlineKeyboardMarkup:
for post_id, post in POSTS.items():
markup.add(
types.InlineKeyboardButton(
post["title"], callback_data=posts_cb.new(id=post_id, action="view")
)
post['title'],
callback_data=posts_cb.new(id=post_id, action='view')),
)
return markup
def format_post(post_id: str, post: dict) -> (str, types.InlineKeyboardMarkup):
text = (
f"{md.hbold(post['title'])}\n"
f"{md.quote_html(post['body'])}\n"
f"\n"
f"Votes: {post['votes']}"
text = md.text(
md.hbold(post['title']),
md.quote_html(post['body']),
'', # just new empty line
f"Votes: {post['votes']}",
sep = '\n',
)
markup = types.InlineKeyboardMarkup()
markup.row(
types.InlineKeyboardButton("👍", callback_data=posts_cb.new(id=post_id, action="like")),
types.InlineKeyboardButton("👎", callback_data=posts_cb.new(id=post_id, action="unlike")),
)
markup.add(
types.InlineKeyboardButton("<< Back", callback_data=posts_cb.new(id="-", action="list"))
types.InlineKeyboardButton('👍', callback_data=posts_cb.new(id=post_id, action='like')),
types.InlineKeyboardButton('👎', callback_data=posts_cb.new(id=post_id, action='dislike')),
)
markup.add(types.InlineKeyboardButton('<< Back', callback_data=posts_cb.new(id='-', action='list')))
return text, markup
@dp.message_handler(commands="start")
@dp.message_handler(commands='start')
async def cmd_start(message: types.Message):
await message.reply("Posts", reply_markup=get_keyboard())
await message.reply('Posts', reply_markup=get_keyboard())
@dp.callback_query_handler(posts_cb.filter(action="list"))
@dp.callback_query_handler(posts_cb.filter(action='list'))
async def query_show_list(query: types.CallbackQuery):
await query.message.edit_text("Posts", reply_markup=get_keyboard())
await query.message.edit_text('Posts', reply_markup=get_keyboard())
@dp.callback_query_handler(posts_cb.filter(action="view"))
@dp.callback_query_handler(posts_cb.filter(action='view'))
async def query_view(query: types.CallbackQuery, callback_data: dict):
post_id = callback_data["id"]
post_id = callback_data['id']
post = POSTS.get(post_id, None)
if not post:
return await query.answer("Unknown post!")
return await query.answer('Unknown post!')
text, markup = format_post(post_id, post)
await query.message.edit_text(text, reply_markup=markup)
@dp.callback_query_handler(posts_cb.filter(action=["like", "unlike"]))
@dp.callback_query_handler(posts_cb.filter(action=['like', 'dislike']))
async def query_post_vote(query: types.CallbackQuery, callback_data: dict):
try:
await dp.throttle("vote", rate=1)
await dp.throttle('vote', rate=1)
except Throttled:
return await query.answer("Too many requests.")
return await query.answer('Too many requests.')
post_id = callback_data["id"]
action = callback_data["action"]
post_id = callback_data['id']
action = callback_data['action']
post = POSTS.get(post_id, None)
if not post:
return await query.answer("Unknown post!")
return await query.answer('Unknown post!')
if action == "like":
post["votes"] += 1
elif action == "unlike":
post["votes"] -= 1
if action == 'like':
post['votes'] += 1
elif action == 'dislike':
post['votes'] -= 1
await query.answer("Voted.")
await query.answer('Vote accepted')
text, markup = format_post(post_id, post)
await query.message.edit_text(text, reply_markup=markup)
@ -118,5 +115,5 @@ async def message_not_modified_handler(update, error):
return True
if __name__ == "__main__":
executor.start_polling(dp, loop=loop, skip_updates=True)
if __name__ == '__main__':
executor.start_polling(dp, skip_updates=True)

68
examples/callback_data_factory_simple.py Normal file
View file

@ -0,0 +1,68 @@
"""
This is a simple example of usage of CallbackData factory
For more comprehensive example see callback_data_factory.py
"""
import logging
from aiogram import Bot, Dispatcher, executor, types
from aiogram.contrib.middlewares.logging import LoggingMiddleware
from aiogram.utils.callback_data import CallbackData
from aiogram.utils.exceptions import MessageNotModified
logging.basicConfig(level=logging.INFO)
API_TOKEN = 'BOT_TOKEN_HERE'
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)
dp.middleware.setup(LoggingMiddleware())
vote_cb = CallbackData('vote', 'action') # vote:<action>
likes = {} # user_id: amount_of_likes
def get_keyboard():
return types.InlineKeyboardMarkup().row(
types.InlineKeyboardButton('👍', callback_data=vote_cb.new(action='up')),
types.InlineKeyboardButton('👎', callback_data=vote_cb.new(action='down')),
)
@dp.message_handler(commands=['start'])
async def cmd_start(message: types.Message):
amount_of_likes = likes.get(message.from_user.id, 0) # get value if key exists else set to 0
await message.reply(f'Vote! You have {amount_of_likes} votes now.', reply_markup=get_keyboard())
@dp.callback_query_handler(vote_cb.filter(action=['up', 'down']))
async def callback_vote_action(query: types.CallbackQuery, callback_data: dict):
logging.info('Got this callback data: %r', callback_data) # callback_data contains all info from callback data
await query.answer() # don't forget to answer callback query as soon as possible
callback_data_action = callback_data['action']
likes_count = likes.get(query.from_user.id, 0)
if callback_data_action == 'up':
likes_count += 1
else:
likes_count -= 1
likes[query.from_user.id] = likes_count # update amount of likes in storage
await bot.edit_message_text(
f'You voted {callback_data_action}! Now you have {likes_count} vote[s].',
query.from_user.id,
query.message.message_id,
reply_markup=get_keyboard(),
)
@dp.errors_handler(exception=MessageNotModified) # handle the cases when this exception raises
async def message_not_modified_handler(update, error):
return True
if __name__ == '__main__':
executor.start_polling(dp, skip_updates=True)

29
examples/check_user_language.py
View file

@ -2,17 +2,16 @@
Babel is required.
"""
import asyncio
import logging
from aiogram import Bot, Dispatcher, executor, md, types
API_TOKEN = "BOT TOKEN HERE"
API_TOKEN = 'BOT TOKEN HERE'
logging.basicConfig(level=logging.INFO)
loop = asyncio.get_event_loop()
bot = Bot(token=API_TOKEN, loop=loop, parse_mode=types.ParseMode.MARKDOWN)
bot = Bot(token=API_TOKEN, parse_mode=types.ParseMode.MARKDOWN)
dp = Dispatcher(bot)
@ -20,17 +19,15 @@ dp = Dispatcher(bot)
async def check_language(message: types.Message):
locale = message.from_user.locale
await message.reply(
md.text(
md.bold("Info about your language:"),
md.text(" 🔸", md.bold("Code:"), md.italic(locale.locale)),
md.text(" 🔸", md.bold("Territory:"), md.italic(locale.territory or "Unknown")),
md.text(" 🔸", md.bold("Language name:"), md.italic(locale.language_name)),
md.text(" 🔸", md.bold("English language name:"), md.italic(locale.english_name)),
sep="\n",
)
)
await message.reply(md.text(
md.bold('Info about your language:'),
md.text('🔸', md.bold('Code:'), md.code(locale.language)),
md.text('🔸', md.bold('Territory:'), md.code(locale.territory or 'Unknown')),
md.text('🔸', md.bold('Language name:'), md.code(locale.language_name)),
md.text('🔸', md.bold('English language name:'), md.code(locale.english_name)),
sep='\n',
))
if __name__ == "__main__":
executor.start_polling(dp, loop=loop, skip_updates=True)
if __name__ == '__main__':
executor.start_polling(dp, skip_updates=True)

24
examples/echo_bot.py
View file

@ -7,7 +7,7 @@ import logging
from aiogram import Bot, Dispatcher, executor, types
API_TOKEN = "BOT TOKEN HERE"
API_TOKEN = 'BOT TOKEN HERE'
# Configure logging
logging.basicConfig(level=logging.INFO)
@ -17,29 +17,37 @@ bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)
@dp.message_handler(commands=["start", "help"])
@dp.message_handler(commands=['start', 'help'])
async def send_welcome(message: types.Message):
"""
This handler will be called when client send `/start` or `/help` commands.
This handler will be called when user sends `/start` or `/help` command
"""
await message.reply("Hi!\nI'm EchoBot!\nPowered by aiogram.")
@dp.message_handler(regexp="(^cat[s]?$|puss)")
@dp.message_handler(regexp='(^cat[s]?$|puss)')
async def cats(message: types.Message):
with open("data/cats.jpg", "rb") as photo:
with open('data/cats.jpg', 'rb') as photo:
'''
# Old fashioned way:
await bot.send_photo(
message.chat.id,
photo,
caption="Cats is here 😺",
caption='Cats are here 😺',
reply_to_message_id=message.message_id,
)
'''
await message.reply_photo(photo, caption='Cats are here 😺')
@dp.message_handler()
async def echo(message: types.Message):
await bot.send_message(message.chat.id, message.text)
# old style:
# await bot.send_message(message.chat.id, message.text)
await message.reply(message.text, reply=False)
if __name__ == "__main__":
if __name__ == '__main__':
executor.start_polling(dp, skip_updates=True)

57
examples/finite_state_machine_example.py
View file

@ -1,19 +1,20 @@
import asyncio
from typing import Optional
import logging
import aiogram.utils.markdown as md
from aiogram import Bot, Dispatcher, types
from aiogram.contrib.fsm_storage.memory import MemoryStorage
from aiogram.dispatcher import FSMContext
from aiogram.dispatcher.filters import Text
from aiogram.dispatcher.filters.state import State, StatesGroup
from aiogram.types import ParseMode
from aiogram.utils import executor
API_TOKEN = "BOT TOKEN HERE"
logging.basicConfig(level=logging.INFO)
loop = asyncio.get_event_loop()
API_TOKEN = 'BOT TOKEN HERE'
bot = Bot(token=API_TOKEN, loop=loop)
bot = Bot(token=API_TOKEN)
# For example use simple MemoryStorage for Dispatcher.
storage = MemoryStorage()
@ -27,7 +28,7 @@ class Form(StatesGroup):
gender = State() # Will be represented in storage as 'Form:gender'
@dp.message_handler(commands=["start"])
@dp.message_handler(commands='start')
async def cmd_start(message: types.Message):
"""
Conversation's entry point
@ -39,21 +40,21 @@ async def cmd_start(message: types.Message):
# You can use state '*' if you need to handle all states
@dp.message_handler(state="*", commands=["cancel"])
@dp.message_handler(lambda message: message.text.lower() == "cancel", state="*")
async def cancel_handler(
message: types.Message, state: FSMContext, raw_state: Optional[str] = None
):
@dp.message_handler(state='*', commands='cancel')
@dp.message_handler(Text(equals='cancel', ignore_case=True), state='*')
async def cancel_handler(message: types.Message, state: FSMContext):
"""
Allow user to cancel any action
"""
if raw_state is None:
current_state = await state.get_state()
if current_state is None:
return
logging.info('Cancelling state %r', current_state)
# Cancel state and inform user about it
await state.finish()
# And remove keyboard (just in case)
await message.reply("Canceled.", reply_markup=types.ReplyKeyboardRemove())
await message.reply('Cancelled.', reply_markup=types.ReplyKeyboardRemove())
@dp.message_handler(state=Form.name)
@ -62,7 +63,7 @@ async def process_name(message: types.Message, state: FSMContext):
Process user name
"""
async with state.proxy() as data:
data["name"] = message.text
data['name'] = message.text
await Form.next()
await message.reply("How old are you?")
@ -70,7 +71,7 @@ async def process_name(message: types.Message, state: FSMContext):
# Check age. Age gotta be digit
@dp.message_handler(lambda message: not message.text.isdigit(), state=Form.age)
async def failed_process_age(message: types.Message):
async def process_age_invalid(message: types.Message):
"""
If age is invalid
"""
@ -91,20 +92,18 @@ async def process_age(message: types.Message, state: FSMContext):
await message.reply("What is your gender?", reply_markup=markup)
@dp.message_handler(
lambda message: message.text not in ["Male", "Female", "Other"], state=Form.gender
)
async def failed_process_gender(message: types.Message):
@dp.message_handler(lambda message: message.text not in ["Male", "Female", "Other"], state=Form.gender)
async def process_gender_invalid(message: types.Message):
"""
In this example gender has to be one of: Male, Female, Other.
"""
return await message.reply("Bad gender name. Choose you gender from keyboard.")
return await message.reply("Bad gender name. Choose your gender from the keyboard.")
@dp.message_handler(state=Form.gender)
async def process_gender(message: types.Message, state: FSMContext):
async with state.proxy() as data:
data["gender"] = message.text
data['gender'] = message.text
# Remove keyboard
markup = types.ReplyKeyboardRemove()
@ -113,18 +112,18 @@ async def process_gender(message: types.Message, state: FSMContext):
await bot.send_message(
message.chat.id,
md.text(
md.text("Hi! Nice to meet you,", md.bold(data["name"])),
md.text("Age:", data["age"]),
md.text("Gender:", data["gender"]),
sep="\n",
md.text('Hi! Nice to meet you,', md.bold(data['name'])),
md.text('Age:', md.code(data['age'])),
md.text('Gender:', data['gender']),
sep='\n',
),
reply_markup=markup,
parse_mode=ParseMode.MARKDOWN,
)
# Finish conversation
data.state = None
# Finish conversation
await state.finish()
if __name__ == "__main__":
executor.start_polling(dp, loop=loop, skip_updates=True)
if __name__ == '__main__':
executor.start_polling(dp, skip_updates=True)

58
examples/i18n_example.py
View file

@ -3,6 +3,19 @@ Internalize your bot
Step 1: extract texts
# pybabel extract i18n_example.py -o locales/mybot.pot
Some useful options:
- Extract texts with pluralization support
# -k __:1,2
- Add comments for translators, you can use another tag if you want (TR)
# --add-comments=NOTE
- Disable comments with string location in code
# --no-location
- Set project name
# --project=MySuperBot
- Set version
# --version=2.2
Step 2: create *.po files. For e.g. create en, ru, uk locales.
# echo {en,ru,uk} | xargs -n1 pybabel init -i locales/mybot.pot -d locales -D mybot -l
Step 3: translate texts
@ -24,11 +37,11 @@ from pathlib import Path
from aiogram import Bot, Dispatcher, executor, types
from aiogram.contrib.middlewares.i18n import I18nMiddleware
TOKEN = "BOT TOKEN HERE"
I18N_DOMAIN = "mybot"
TOKEN = 'BOT_TOKEN_HERE'
I18N_DOMAIN = 'mybot'
BASE_DIR = Path(__file__).parent
LOCALES_DIR = BASE_DIR / "locales"
LOCALES_DIR = BASE_DIR / 'locales'
bot = Bot(TOKEN, parse_mode=types.ParseMode.HTML)
dp = Dispatcher(bot)
@ -41,16 +54,45 @@ dp.middleware.setup(i18n)
_ = i18n.gettext
@dp.message_handler(commands=["start"])
@dp.message_handler(commands='start')
async def cmd_start(message: types.Message):
# Simply use `_('message')` instead of `'message'` and never use f-strings for translatable texts.
await message.reply(_("Hello, <b>{user}</b>!").format(user=message.from_user.full_name))
await message.reply(_('Hello, <b>{user}</b>!').format(user=message.from_user.full_name))
@dp.message_handler(commands=["lang"])
@dp.message_handler(commands='lang')
async def cmd_lang(message: types.Message, locale):
await message.reply(_("Your current language: <i>{language}</i>").format(language=locale))
# For setting custom lang you have to modify i18n middleware, like this:
# https://github.com/aiogram/EventsTrackerBot/blob/master/modules/base/middlewares.py
await message.reply(_('Your current language: <i>{language}</i>').format(language=locale))
# If you care about pluralization, here's small handler
# And also, there's and example of comments for translators. Most translation tools support them.
# Alias for gettext method, parser will understand double underscore as plural (aka ngettext)
__ = i18n.gettext
if __name__ == "__main__":
# some likes manager
LIKES_STORAGE = {'count': 0}
def get_likes() -> int:
return LIKES_STORAGE['count']
def increase_likes() -> int:
LIKES_STORAGE['count'] += 1
return get_likes()
#
@dp.message_handler(commands='like')
async def cmd_like(message: types.Message, locale):
likes = increase_likes()
# NOTE: This is comment for a translator
await message.reply(__('Aiogram has {number} like!', 'Aiogram has {number} likes!', likes).format(number=likes))
if __name__ == '__main__':
executor.start_polling(dp, skip_updates=True)

36
examples/id_filter_example.py Normal file
View file

@ -0,0 +1,36 @@
from aiogram import Bot, Dispatcher, executor, types
from aiogram.dispatcher.handler import SkipHandler
API_TOKEN = 'BOT_TOKEN_HERE'
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)
user_id_required = None # TODO: Set id here
chat_id_required = user_id_required # Change for use in groups (user_id == chat_id in pm)
@dp.message_handler(user_id=user_id_required)
async def handler1(msg: types.Message):
await bot.send_message(msg.chat.id, "Hello, checking with user_id=")
raise SkipHandler # just for demo
@dp.message_handler(chat_id=chat_id_required)
async def handler2(msg: types.Message):
await bot.send_message(msg.chat.id, "Hello, checking with chat_id=")
raise SkipHandler # just for demo
@dp.message_handler(user_id=user_id_required, chat_id=chat_id_required)
async def handler3(msg: types.Message):
await msg.reply("Hello from user= & chat_id=", reply=False)
@dp.message_handler(user_id=[user_id_required, 42]) # TODO: You can add any number of ids here
async def handler4(msg: types.Message):
await msg.reply("Checked user_id with list!", reply=False)
if __name__ == '__main__':
executor.start_polling(dp)

34
examples/inline_bot.py
View file

@ -1,25 +1,37 @@
import asyncio
import hashlib
import logging
from aiogram import Bot, types, Dispatcher, executor
from aiogram import Bot, Dispatcher, executor
from aiogram.types import InlineQuery, \
InputTextMessageContent, InlineQueryResultArticle
API_TOKEN = "BOT TOKEN HERE"
API_TOKEN = 'BOT_TOKEN_HERE'
logging.basicConfig(level=logging.DEBUG)
loop = asyncio.get_event_loop()
bot = Bot(token=API_TOKEN, loop=loop)
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)
@dp.inline_handler()
async def inline_echo(inline_query: types.InlineQuery):
input_content = types.InputTextMessageContent(inline_query.query or "echo")
item = types.InlineQueryResultArticle(
id="1", title="echo", input_message_content=input_content
async def inline_echo(inline_query: InlineQuery):
# id affects both preview and content,
# so it has to be unique for each result
# (Unique identifier for this result, 1-64 Bytes)
# you can set your unique id's
# but for example i'll generate it based on text because I know, that
# only text will be passed in this example
text = inline_query.query or 'echo'
input_content = InputTextMessageContent(text)
result_id: str = hashlib.md5(text.encode()).hexdigest()
item = InlineQueryResultArticle(
id=result_id,
title=f'Result {text!r}',
input_message_content=input_content,
)
# don't forget to set cache_time=1 for testing (default is 300s or 5m)
await bot.answer_inline_query(inline_query.id, results=[item], cache_time=1)
if __name__ == "__main__":
executor.start_polling(dp, loop=loop, skip_updates=True)
if __name__ == '__main__':
executor.start_polling(dp, skip_updates=True)

62
examples/inline_keyboard_example.py Normal file
View file

@ -0,0 +1,62 @@
"""
This bot is created for the demonstration of a usage of inline keyboards.
"""
import logging
from aiogram import Bot, Dispatcher, executor, types
API_TOKEN = 'BOT_TOKEN_HERE'
# Configure logging
logging.basicConfig(level=logging.INFO)
# Initialize bot and dispatcher
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)
@dp.message_handler(commands='start')
async def start_cmd_handler(message: types.Message):
keyboard_markup = types.InlineKeyboardMarkup(row_width=3)
# default row_width is 3, so here we can omit it actually
# kept for clearness
text_and_data = (
('Yes!', 'yes'),
('No!', 'no'),
)
# in real life for the callback_data the callback data factory should be used
# here the raw string is used for the simplicity
row_btns = (types.InlineKeyboardButton(text, callback_data=data) for text, data in text_and_data)
keyboard_markup.row(*row_btns)
keyboard_markup.add(
# url buttons have no callback data
types.InlineKeyboardButton('aiogram source', url='https://github.com/aiogram/aiogram'),
)
await message.reply("Hi!\nDo you love aiogram?", reply_markup=keyboard_markup)
# Use multiple registrators. Handler will execute when one of the filters is OK
@dp.callback_query_handler(text='no') # if cb.data == 'no'
@dp.callback_query_handler(text='yes') # if cb.data == 'yes'
async def inline_kb_answer_callback_handler(query: types.CallbackQuery):
answer_data = query.data
# always answer callback queries, even if you have nothing to say
await query.answer(f'You answered with {answer_data!r}')
if answer_data == 'yes':
text = 'Great, me too!'
elif answer_data == 'no':
text = 'Oh no...Why so?'
else:
text = f'Unexpected callback data {answer_data!r}!'
await bot.send_message(query.from_user.id, text)
if __name__ == '__main__':
executor.start_polling(dp, skip_updates=True)

13
examples/locales/mybot.pot
View file

@ -1,27 +1,26 @@
# Translations template for PROJECT.
# Copyright (C) 2018 ORGANIZATION
# Copyright (C) 2019 ORGANIZATION
# This file is distributed under the same license as the PROJECT project.
# FIRST AUTHOR <EMAIL@ADDRESS>, 2018.
# FIRST AUTHOR <EMAIL@ADDRESS>, 2019.
#
#, fuzzy
msgid ""
msgstr ""
"Project-Id-Version: PROJECT VERSION\n"
"Report-Msgid-Bugs-To: EMAIL@ADDRESS\n"
"POT-Creation-Date: 2018-06-30 03:50+0300\n"
"POT-Creation-Date: 2019-08-10 17:51+0300\n"
"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language-Team: LANGUAGE <LL@li.org>\n"
"MIME-Version: 1.0\n"
"Content-Type: text/plain; charset=utf-8\n"
"Content-Transfer-Encoding: 8bit\n"
"Generated-By: Babel 2.6.0\n"
"Generated-By: Babel 2.7.0\n"
#: i18n_example.py:48
#: i18n_example.py:60
msgid "Hello, <b>{user}</b>!"
msgstr ""
#: i18n_example.py:53
#: i18n_example.py:67
msgid "Your current language: <i>{language}</i>"
msgstr ""

20
examples/locales/ru/LC_MESSAGES/mybot.po
View file

@ -1,14 +1,14 @@
# Russian translations for PROJECT.
# Copyright (C) 2018 ORGANIZATION
# Copyright (C) 2019 ORGANIZATION
# This file is distributed under the same license as the PROJECT project.
# FIRST AUTHOR <EMAIL@ADDRESS>, 2018.
# FIRST AUTHOR <EMAIL@ADDRESS>, 2019.
#
msgid ""
msgstr ""
"Project-Id-Version: PROJECT VERSION\n"
"Report-Msgid-Bugs-To: EMAIL@ADDRESS\n"
"POT-Creation-Date: 2018-06-30 03:50+0300\n"
"PO-Revision-Date: 2018-06-30 03:43+0300\n"
"POT-Creation-Date: 2019-08-10 17:51+0300\n"
"PO-Revision-Date: 2019-08-10 17:52+0300\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language: ru\n"
"Language-Team: ru <LL@li.org>\n"
@ -17,13 +17,19 @@ msgstr ""
"MIME-Version: 1.0\n"
"Content-Type: text/plain; charset=utf-8\n"
"Content-Transfer-Encoding: 8bit\n"
"Generated-By: Babel 2.6.0\n"
"Generated-By: Babel 2.7.0\n"
#: i18n_example.py:48
#: i18n_example.py:60
msgid "Hello, <b>{user}</b>!"
msgstr "Привет, <b>{user}</b>!"
#: i18n_example.py:53
#: i18n_example.py:67
msgid "Your current language: <i>{language}</i>"
msgstr "Твой язык: <i>{language}</i>"
#: i18n_example.py:95
msgid "Aiogram has {number} like!"
msgid_plural "Aiogram has {number} likes!"
msgstr[0] "Aiogram имеет {number} лайк!"
msgstr[1] "Aiogram имеет {number} лайка!"
msgstr[2] "Aiogram имеет {number} лайков!"

20
examples/media_group.py
View file

@ -2,10 +2,10 @@ import asyncio
from aiogram import Bot, Dispatcher, executor, filters, types
API_TOKEN = "BOT TOKEN HERE"
loop = asyncio.get_event_loop()
bot = Bot(token=API_TOKEN, loop=loop)
API_TOKEN = 'BOT_TOKEN_HERE'
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)
@ -14,23 +14,23 @@ async def send_welcome(message: types.Message):
# So... At first I want to send something like this:
await message.reply("Do you want to see many pussies? Are you ready?")
# And wait few seconds...
# Wait a little...
await asyncio.sleep(1)
# Good bots should send chat actions. Or not.
# Good bots should send chat actions...
await types.ChatActions.upload_photo()
# Create media group
media = types.MediaGroup()
# Attach local file
media.attach_photo(types.InputFile("data/cat.jpg"), "Cat!")
media.attach_photo(types.InputFile('data/cat.jpg'), 'Cat!')
# More local files and more cats!
media.attach_photo(types.InputFile("data/cats.jpg"), "More cats!")
media.attach_photo(types.InputFile('data/cats.jpg'), 'More cats!')
# You can also use URL's
# For example: get random puss:
media.attach_photo("http://lorempixel.com/400/200/cats/", "Random cat.")
media.attach_photo('http://lorempixel.com/400/200/cats/', 'Random cat.')
# And you can also use file ID:
# media.attach_photo('<file_id>', 'cat-cat-cat.')
@ -39,5 +39,5 @@ async def send_welcome(message: types.Message):
await message.reply_media_group(media=media)
if __name__ == "__main__":
executor.start_polling(dp, loop=loop, skip_updates=True)
if __name__ == '__main__':
executor.start_polling(dp, skip_updates=True)

34
examples/middleware_and_antiflood.py
View file

@ -7,14 +7,12 @@ from aiogram.dispatcher.handler import CancelHandler, current_handler
from aiogram.dispatcher.middlewares import BaseMiddleware
from aiogram.utils.exceptions import Throttled
TOKEN = "BOT TOKEN HERE"
loop = asyncio.get_event_loop()
TOKEN = 'BOT_TOKEN_HERE'
# In this example Redis storage is used
storage = RedisStorage2(db=5)
bot = Bot(token=TOKEN, loop=loop)
bot = Bot(token=TOKEN)
dp = Dispatcher(bot, storage=storage)
@ -28,9 +26,9 @@ def rate_limit(limit: int, key=None):
"""
def decorator(func):
setattr(func, "throttling_rate_limit", limit)
setattr(func, 'throttling_rate_limit', limit)
if key:
setattr(func, "throttling_key", key)
setattr(func, 'throttling_key', key)
return func
return decorator
@ -41,7 +39,7 @@ class ThrottlingMiddleware(BaseMiddleware):
Simple middleware
"""
def __init__(self, limit=DEFAULT_RATE_LIMIT, key_prefix="antiflood_"):
def __init__(self, limit=DEFAULT_RATE_LIMIT, key_prefix='antiflood_'):
self.rate_limit = limit
self.prefix = key_prefix
super(ThrottlingMiddleware, self).__init__()
@ -59,8 +57,8 @@ class ThrottlingMiddleware(BaseMiddleware):
dispatcher = Dispatcher.get_current()
# If handler was configured, get rate limit and key from handler
if handler:
limit = getattr(handler, "throttling_rate_limit", self.rate_limit)
key = getattr(handler, "throttling_key", f"{self.prefix}_{handler.__name__}")
limit = getattr(handler, 'throttling_rate_limit', self.rate_limit)
key = getattr(handler, 'throttling_key', f"{self.prefix}_{handler.__name__}")
else:
limit = self.rate_limit
key = f"{self.prefix}_message"
@ -85,7 +83,7 @@ class ThrottlingMiddleware(BaseMiddleware):
handler = current_handler.get()
dispatcher = Dispatcher.get_current()
if handler:
key = getattr(handler, "throttling_key", f"{self.prefix}_{handler.__name__}")
key = getattr(handler, 'throttling_key', f"{self.prefix}_{handler.__name__}")
else:
key = f"{self.prefix}_message"
@ -94,7 +92,7 @@ class ThrottlingMiddleware(BaseMiddleware):
# Prevent flooding
if throttled.exceeded_count <= 2:
await message.reply("Too many requests! ")
await message.reply('Too many requests! ')
# Sleep.
await asyncio.sleep(delta)
@ -104,21 +102,19 @@ class ThrottlingMiddleware(BaseMiddleware):
# If current message is not last with current key - do not send message
if thr.exceeded_count == throttled.exceeded_count:
await message.reply("Unlocked.")
await message.reply('Unlocked.')
@dp.message_handler(commands=["start"])
@rate_limit(
5, "start"
) # this is not required but you can configure throttling manager for current handler using it
@dp.message_handler(commands=['start'])
@rate_limit(5, 'start') # this is not required but you can configure throttling manager for current handler using it
async def cmd_test(message: types.Message):
# You can use this command every 5 seconds
await message.reply("Test passed! You can use this command every 5 seconds.")
await message.reply('Test passed! You can use this command every 5 seconds.')
if __name__ == "__main__":
if __name__ == '__main__':
# Setup middleware
dp.middleware.setup(ThrottlingMiddleware())
# Start long-polling
executor.start_polling(dp, loop=loop)
executor.start_polling(dp)

144
examples/payments.py
View file

@ -1,121 +1,97 @@
import asyncio
from aiogram import Bot
from aiogram import types
from aiogram.dispatcher import Dispatcher
from aiogram.types.message import ContentTypes
from aiogram.utils import executor
BOT_TOKEN = "BOT TOKEN HERE"
PAYMENTS_PROVIDER_TOKEN = "123456789:TEST:1234567890abcdef1234567890abcdef"
loop = asyncio.get_event_loop()
BOT_TOKEN = 'BOT_TOKEN_HERE'
PAYMENTS_PROVIDER_TOKEN = '123456789:TEST:1422'
bot = Bot(BOT_TOKEN)
dp = Dispatcher(bot, loop=loop)
dp = Dispatcher(bot)
# Setup prices
prices = [
types.LabeledPrice(label="Working Time Machine", amount=5750),
types.LabeledPrice(label="Gift wrapping", amount=500),
types.LabeledPrice(label='Working Time Machine', amount=5750),
types.LabeledPrice(label='Gift wrapping', amount=500),
]
# Setup shipping options
shipping_options = [
types.ShippingOption(id="instant", title="WorldWide Teleporter").add(
types.LabeledPrice("Teleporter", 1000)
),
types.ShippingOption(id="pickup", title="Local pickup").add(types.LabeledPrice("Pickup", 300)),
types.ShippingOption(id='instant', title='WorldWide Teleporter').add(types.LabeledPrice('Teleporter', 1000)),
types.ShippingOption(id='pickup', title='Local pickup').add(types.LabeledPrice('Pickup', 300)),
]
@dp.message_handler(commands=["start"])
@dp.message_handler(commands=['start'])
async def cmd_start(message: types.Message):
await bot.send_message(
message.chat.id,
"Hello, I'm the demo merchant bot."
" I can sell you a Time Machine."
" Use /buy to order one, /terms for Terms and Conditions",
)
await bot.send_message(message.chat.id,
"Hello, I'm the demo merchant bot."
" I can sell you a Time Machine."
" Use /buy to order one, /terms for Terms and Conditions")
@dp.message_handler(commands=["terms"])
@dp.message_handler(commands=['terms'])
async def cmd_terms(message: types.Message):
await bot.send_message(
message.chat.id,
"Thank you for shopping with our demo bot. We hope you like your new time machine!\n"
"1. If your time machine was not delivered on time, please rethink your concept of time"
" and try again.\n"
"2. If you find that your time machine is not working, kindly contact our future service"
" workshops on Trappist-1e. They will be accessible anywhere between"
" May 2075 and November 4000 C.E.\n"
"3. If you would like a refund, kindly apply for one yesterday and we will have sent it"
" to you immediately.",
)
await bot.send_message(message.chat.id,
'Thank you for shopping with our demo bot. We hope you like your new time machine!\n'
'1. If your time machine was not delivered on time, please rethink your concept of time'
' and try again.\n'
'2. If you find that your time machine is not working, kindly contact our future service'
' workshops on Trappist-1e. They will be accessible anywhere between'
' May 2075 and November 4000 C.E.\n'
'3. If you would like a refund, kindly apply for one yesterday and we will have sent it'
' to you immediately.')
@dp.message_handler(commands=["buy"])
@dp.message_handler(commands=['buy'])
async def cmd_buy(message: types.Message):
await bot.send_message(
message.chat.id,
"Real cards won't work with me, no money will be debited from your account."
" Use this test card number to pay for your Time Machine: `4242 4242 4242 4242`"
"\n\nThis is your demo invoice:",
parse_mode="Markdown",
)
await bot.send_invoice(
message.chat.id,
title="Working Time Machine",
description="Want to visit your great-great-great-grandparents?"
" Make a fortune at the races?"
" Shake hands with Hammurabi and take a stroll in the Hanging Gardens?"
" Order our Working Time Machine today!",
provider_token=PAYMENTS_PROVIDER_TOKEN,
currency="usd",
photo_url="https://images.fineartamerica.com/images-medium-large/2-the-time-machine-dmitriy-khristenko.jpg",
photo_height=512, # !=0/None or picture won't be shown
photo_width=512,
photo_size=512,
is_flexible=True, # True If you need to set up Shipping Fee
prices=prices,
start_parameter="time-machine-example",
payload="HAPPY FRIDAYS COUPON",
)
await bot.send_message(message.chat.id,
"Real cards won't work with me, no money will be debited from your account."
" Use this test card number to pay for your Time Machine: `4242 4242 4242 4242`"
"\n\nThis is your demo invoice:", parse_mode='Markdown')
await bot.send_invoice(message.chat.id, title='Working Time Machine',
description='Want to visit your great-great-great-grandparents?'
' Make a fortune at the races?'
' Shake hands with Hammurabi and take a stroll in the Hanging Gardens?'
' Order our Working Time Machine today!',
provider_token=PAYMENTS_PROVIDER_TOKEN,
currency='usd',
photo_url='https://telegra.ph/file/d08ff863531f10bf2ea4b.jpg',
photo_height=512, # !=0/None or picture won't be shown
photo_width=512,
photo_size=512,
is_flexible=True, # True If you need to set up Shipping Fee
prices=prices,
start_parameter='time-machine-example',
payload='HAPPY FRIDAYS COUPON')
@dp.shipping_query_handler(func=lambda query: True)
@dp.shipping_query_handler(lambda query: True)
async def shipping(shipping_query: types.ShippingQuery):
await bot.answer_shipping_query(
shipping_query.id,
ok=True,
shipping_options=shipping_options,
error_message="Oh, seems like our Dog couriers are having a lunch right now."
" Try again later!",
)
await bot.answer_shipping_query(shipping_query.id, ok=True, shipping_options=shipping_options,
error_message='Oh, seems like our Dog couriers are having a lunch right now.'
' Try again later!')
@dp.pre_checkout_query_handler(func=lambda query: True)
@dp.pre_checkout_query_handler(lambda query: True)
async def checkout(pre_checkout_query: types.PreCheckoutQuery):
await bot.answer_pre_checkout_query(
pre_checkout_query.id,
ok=True,
error_message="Aliens tried to steal your card's CVV,"
" but we successfully protected your credentials,"
" try to pay again in a few minutes, we need a small rest.",
)
await bot.answer_pre_checkout_query(pre_checkout_query.id, ok=True,
error_message="Aliens tried to steal your card's CVV,"
" but we successfully protected your credentials,"
" try to pay again in a few minutes, we need a small rest.")
@dp.message_handler(content_types=ContentTypes.SUCCESSFUL_PAYMENT)
async def got_payment(message: types.Message):
await bot.send_message(
message.chat.id,
"Hoooooray! Thanks for payment! We will proceed your order for `{} {}`"
" as fast as possible! Stay in touch."
"\n\nUse /buy again to get a Time Machine for your friend!".format(
message.successful_payment.total_amount / 100, message.successful_payment.currency
),
parse_mode="Markdown",
)
await bot.send_message(message.chat.id,
'Hoooooray! Thanks for payment! We will proceed your order for `{} {}`'
' as fast as possible! Stay in touch.'
'\n\nUse /buy again to get a Time Machine for your friend!'.format(
message.successful_payment.total_amount / 100, message.successful_payment.currency),
parse_mode='Markdown')
if __name__ == "__main__":
executor.start_polling(dp, loop=loop)
if __name__ == '__main__':
executor.start_polling(dp, skip_updates=True)

50
examples/proxy_and_emojize.py
View file

@ -1,4 +1,3 @@
import asyncio
import logging
import aiohttp
@ -11,49 +10,52 @@ from aiogram.utils.executor import start_polling
from aiogram.utils.markdown import bold, code, italic, text
# Configure bot here
API_TOKEN = "BOT TOKEN HERE"
PROXY_URL = "http://PROXY_URL" # Or 'socks5://...'
API_TOKEN = 'BOT_TOKEN_HERE'
PROXY_URL = 'http://PROXY_URL' # Or 'socks5://host:port'
# If authentication is required in your proxy then uncomment next line and change login/password for it
# NOTE: If authentication is required in your proxy then uncomment next line and change login/password for it
# PROXY_AUTH = aiohttp.BasicAuth(login='login', password='password')
# And add `proxy_auth=PROXY_AUTH` argument in line 25, like this:
# >>> bot = Bot(token=API_TOKEN, loop=loop, proxy=PROXY_URL, proxy_auth=PROXY_AUTH)
# And add `proxy_auth=PROXY_AUTH` argument in line 30, like this:
# >>> bot = Bot(token=API_TOKEN, proxy=PROXY_URL, proxy_auth=PROXY_AUTH)
# Also you can use Socks5 proxy but you need manually install aiohttp_socks package.
# Get my ip URL
GET_IP_URL = "http://bot.whatismyipaddress.com/"
GET_IP_URL = 'http://bot.whatismyipaddress.com/'
logging.basicConfig(level=logging.INFO)
loop = asyncio.get_event_loop()
bot = Bot(token=API_TOKEN, loop=loop, proxy=PROXY_URL)
bot = Bot(token=API_TOKEN, proxy=PROXY_URL)
# If auth is required:
# bot = Bot(token=API_TOKEN, proxy=PROXY_URL, proxy_auth=PROXY_AUTH)
dp = Dispatcher(bot)
async def fetch(url, proxy=None, proxy_auth=None):
async with aiohttp.ClientSession() as session:
async with session.get(url, proxy=proxy, proxy_auth=proxy_auth) as response:
return await response.text()
async def fetch(url, session):
async with session.get(url) as response:
return await response.text()
@dp.message_handler(commands=["start"])
@dp.message_handler(commands=['start'])
async def cmd_start(message: types.Message):
# fetching urls will take some time, so notify user that everything is OK
await types.ChatActions.typing()
content = []
# Make request (without proxy)
ip = await fetch(GET_IP_URL)
content.append(text(":globe_showing_Americas:", bold("IP:"), code(ip)))
async with aiohttp.ClientSession() as session:
ip = await fetch(GET_IP_URL, session)
content.append(text(':globe_showing_Americas:', bold('IP:'), code(ip)))
# This line is formatted to '🌎 *IP:* `YOUR IP`'
# Make request through proxy
ip = await fetch(GET_IP_URL, bot.proxy, bot.proxy_auth)
content.append(text(":locked_with_key:", bold("IP:"), code(ip), italic("via proxy")))
# Make request through bot's proxy
ip = await fetch(GET_IP_URL, bot.session)
content.append(text(':locked_with_key:', bold('IP:'), code(ip), italic('via proxy')))
# This line is formatted to '🔐 *IP:* `YOUR IP` _via proxy_'
# Send content
await bot.send_message(
message.chat.id, emojize(text(*content, sep="\n")), parse_mode=ParseMode.MARKDOWN
)
await bot.send_message(message.chat.id, emojize(text(*content, sep='\n')), parse_mode=ParseMode.MARKDOWN)
# In this example you can see emoji codes: ":globe_showing_Americas:" and ":locked_with_key:"
# You can find full emoji cheat sheet at https://www.webpagefx.com/tools/emoji-cheat-sheet/
@ -63,5 +65,5 @@ async def cmd_start(message: types.Message):
# For example emojize('Moon face :new_moon_face:') is transformed to 'Moon face 🌚'
if __name__ == "__main__":
start_polling(dp, loop=loop, skip_updates=True)
if __name__ == '__main__':
start_polling(dp, skip_updates=True)

24
examples/regexp_commands_filter_example.py
View file

@ -2,16 +2,28 @@ from aiogram import Bot, types
from aiogram.dispatcher import Dispatcher, filters
from aiogram.utils import executor
bot = Bot(token="TOKEN")
bot = Bot(token='BOT_TOKEN_HERE', parse_mode=types.ParseMode.HTML)
dp = Dispatcher(bot)
@dp.message_handler(filters.RegexpCommandsFilter(regexp_commands=["item_([0-9]*)"]))
@dp.message_handler(filters.RegexpCommandsFilter(regexp_commands=['item_([0-9]*)']))
async def send_welcome(message: types.Message, regexp_command):
await message.reply(
"You have requested an item with number: {}".format(regexp_command.group(1))
await message.reply(f"You have requested an item with id <code>{regexp_command.group(1)}</code>")
@dp.message_handler(commands='start')
async def create_deeplink(message: types.Message):
bot_user = await bot.me
bot_username = bot_user.username
deeplink = f'https://t.me/{bot_username}?start=item_12345'
text = (
f'Either send a command /item_1234 or follow this link {deeplink} and then click start\n'
'It also can be hidden in a inline button\n\n'
'Or just send <code>/start item_123</code>'
)
await message.reply(text, disable_web_page_preview=True)
if __name__ == "__main__":
executor.start_polling(dp)
if __name__ == '__main__':
executor.start_polling(dp, skip_updates=True)

67
examples/regular_keyboard_example.py Normal file
View file

@ -0,0 +1,67 @@
"""
This bot is created for the demonstration of a usage of regular keyboards.
"""
import logging
from aiogram import Bot, Dispatcher, executor, types
API_TOKEN = 'BOT_TOKEN_HERE'
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
# Initialize bot and dispatcher
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)
@dp.message_handler(commands='start')
async def start_cmd_handler(message: types.Message):
keyboard_markup = types.ReplyKeyboardMarkup(row_width=3)
# default row_width is 3, so here we can omit it actually
# kept for clearness
btns_text = ('Yes!', 'No!')
keyboard_markup.row(*(types.KeyboardButton(text) for text in btns_text))
# adds buttons as a new row to the existing keyboard
# the behaviour doesn't depend on row_width attribute
more_btns_text = (
"I don't know",
"Who am i?",
"Where am i?",
"Who is there?",
)
keyboard_markup.add(*(types.KeyboardButton(text) for text in more_btns_text))
# adds buttons. New rows are formed according to row_width parameter
await message.reply("Hi!\nDo you like aiogram?", reply_markup=keyboard_markup)
@dp.message_handler()
async def all_msg_handler(message: types.Message):
# pressing of a KeyboardButton is the same as sending the regular message with the same text
# so, to handle the responses from the keyboard, we need to use a message_handler
# in real bot, it's better to define message_handler(text="...") for each button
# but here for the simplicity only one handler is defined
button_text = message.text
logger.debug('The answer is %r', button_text) # print the text we've got
if button_text == 'Yes!':
reply_text = "That's great"
elif button_text == 'No!':
reply_text = "Oh no! Why?"
else:
reply_text = "Keep calm...Everything is fine"
await message.reply(reply_text, reply_markup=types.ReplyKeyboardRemove())
# with message, we send types.ReplyKeyboardRemove() to hide the keyboard
if __name__ == '__main__':
executor.start_polling(dp, skip_updates=True)

53
examples/text_filter_example.py Normal file
View file

@ -0,0 +1,53 @@
"""
This is a bot to show the usage of the builtin Text filter
Instead of a list, a single element can be passed to any filter, it will be treated as list with an element
"""
import logging
from aiogram import Bot, Dispatcher, executor, types
API_TOKEN = 'BOT_TOKEN_HERE'
# Configure logging
logging.basicConfig(level=logging.INFO)
# Initialize bot and dispatcher
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)
# if the text from user in the list
@dp.message_handler(text=['text1', 'text2'])
async def text_in_handler(message: types.Message):
await message.answer("The message text equals to one of in the list!")
# if the text contains any string
@dp.message_handler(text_contains='example1')
@dp.message_handler(text_contains='example2')
async def text_contains_any_handler(message: types.Message):
await message.answer("The message text contains any of strings")
# if the text contains all the strings from the list
@dp.message_handler(text_contains=['str1', 'str2'])
async def text_contains_all_handler(message: types.Message):
await message.answer("The message text contains all strings from the list")
# if the text starts with any string from the list
@dp.message_handler(text_startswith=['prefix1', 'prefix2'])
async def text_startswith_handler(message: types.Message):
await message.answer("The message text starts with any of prefixes")
# if the text ends with any string from the list
@dp.message_handler(text_endswith=['postfix1', 'postfix2'])
async def text_endswith_handler(message: types.Message):
await message.answer("The message text ends with any of postfixes")
if __name__ == '__main__':
executor.start_polling(dp, skip_updates=True)

43
examples/throtling_example.py
View file

@ -1,43 +0,0 @@
"""
Example for throttling manager.
You can use that for flood controlling.
"""
import asyncio
import logging
from aiogram import Bot, types
from aiogram.contrib.fsm_storage.memory import MemoryStorage
from aiogram.dispatcher import Dispatcher
from aiogram.utils.exceptions import Throttled
from aiogram.utils.executor import start_polling
API_TOKEN = "BOT TOKEN HERE"
logging.basicConfig(level=logging.INFO)
loop = asyncio.get_event_loop()
bot = Bot(token=API_TOKEN, loop=loop)
# Throttling manager does not work without Leaky Bucket.
# Then need to use storages. For example use simple in-memory storage.
storage = MemoryStorage()
dp = Dispatcher(bot, storage=storage)
@dp.message_handler(commands=["start", "help"])
async def send_welcome(message: types.Message):
try:
# Execute throttling manager with rate-limit equal to 2 seconds for key "start"
await dp.throttle("start", rate=2)
except Throttled:
# If request is throttled, the `Throttled` exception will be raised
await message.reply("Too many requests!")
else:
# Otherwise do something
await message.reply("Hi!\nI'm EchoBot!\nPowered by aiogram.")
if __name__ == "__main__":
start_polling(dp, loop=loop, skip_updates=True)

71
examples/throttling_example.py Normal file
View file

@ -0,0 +1,71 @@
"""
Example for throttling manager.
You can use that for flood controlling.
"""
import logging
from aiogram import Bot, types
from aiogram.contrib.fsm_storage.memory import MemoryStorage
from aiogram.dispatcher import Dispatcher
from aiogram.utils.exceptions import Throttled
from aiogram.utils.executor import start_polling
API_TOKEN = 'BOT_TOKEN_HERE'
logging.basicConfig(level=logging.INFO)
bot = Bot(token=API_TOKEN)
# Throttling manager does not work without Leaky Bucket.
# You need to use a storage. For example use simple in-memory storage.
storage = MemoryStorage()
dp = Dispatcher(bot, storage=storage)
@dp.message_handler(commands=['start'])
async def send_welcome(message: types.Message):
try:
# Execute throttling manager with rate-limit equal to 2 seconds for key "start"
await dp.throttle('start', rate=2)
except Throttled:
# If request is throttled, the `Throttled` exception will be raised
await message.reply('Too many requests!')
else:
# Otherwise do something
await message.reply("Hi!\nI'm EchoBot!\nPowered by aiogram.")
@dp.message_handler(commands=['hi'])
@dp.throttled(lambda msg, loop, *args, **kwargs: loop.create_task(bot.send_message(msg.from_user.id, "Throttled")),
rate=5)
# loop is added to the function to run coroutines from it
async def say_hi(message: types.Message):
await message.answer("Hi")
# the on_throttled object can be either a regular function or coroutine
async def hello_throttled(*args, **kwargs):
# args will be the same as in the original handler
# kwargs will be updated with parameters given to .throttled (rate, key, user_id, chat_id)
print(f"hello_throttled was called with args={args} and kwargs={kwargs}")
message = args[0] # as message was the first argument in the original handler
await message.answer("Throttled")
@dp.message_handler(commands=['hello'])
@dp.throttled(hello_throttled, rate=4)
async def say_hello(message: types.Message):
await message.answer("Hello!")
@dp.message_handler(commands=['help'])
@dp.throttled(rate=5)
# nothing will happen if the handler will be throttled
async def help_handler(message: types.Message):
await message.answer('Help!')
if __name__ == '__main__':
start_polling(dp, skip_updates=True)

215
examples/webhook_example.py
View file

@ -1,199 +1,66 @@
"""
Example outdated
"""
import logging
import asyncio
import ssl
import sys
from aiohttp import web
import aiogram
from aiogram import Bot, types
from aiogram.contrib.fsm_storage.memory import MemoryStorage
from aiogram.contrib.middlewares.logging import LoggingMiddleware
from aiogram.dispatcher import Dispatcher
from aiogram.dispatcher.webhook import get_new_configured_app, SendMessage
from aiogram.types import ChatType, ParseMode, ContentTypes
from aiogram.utils.markdown import hbold, bold, text, link
from aiogram.dispatcher.webhook import SendMessage
from aiogram.utils.executor import start_webhook
TOKEN = "BOT TOKEN HERE"
WEBHOOK_HOST = "example.com" # Domain name or IP addres which your bot is located.
WEBHOOK_PORT = 443 # Telegram Bot API allows only for usage next ports: 443, 80, 88 or 8443
WEBHOOK_URL_PATH = "/webhook" # Part of URL
API_TOKEN = 'BOT_TOKEN_HERE'
# This options needed if you use self-signed SSL certificate
# Instructions: https://core.telegram.org/bots/self-signed
WEBHOOK_SSL_CERT = "./webhook_cert.pem" # Path to the ssl certificate
WEBHOOK_SSL_PRIV = "./webhook_pkey.pem" # Path to the ssl private key
# webhook settings
WEBHOOK_HOST = 'https://your.domain'
WEBHOOK_PATH = '/path/to/api'
WEBHOOK_URL = f"{WEBHOOK_HOST}{WEBHOOK_PATH}"
WEBHOOK_URL = f"https://{WEBHOOK_HOST}:{WEBHOOK_PORT}{WEBHOOK_URL_PATH}"
# Web app settings:
# Use LAN address to listen webhooks
# User any available port in range from 1024 to 49151 if you're using proxy, or WEBHOOK_PORT if you're using direct webhook handling
WEBAPP_HOST = "localhost"
# webserver settings
WEBAPP_HOST = 'localhost' # or ip
WEBAPP_PORT = 3001
BAD_CONTENT = (
ContentTypes.PHOTO & ContentTypes.DOCUMENT & ContentTypes.STICKER & ContentTypes.AUDIO
)
logging.basicConfig(level=logging.INFO)
loop = asyncio.get_event_loop()
bot = Bot(TOKEN, loop=loop)
storage = MemoryStorage()
dp = Dispatcher(bot, storage=storage)
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)
dp.middleware.setup(LoggingMiddleware())
async def cmd_start(message: types.Message):
# Yep. aiogram allows to respond into webhook.
# https://core.telegram.org/bots/api#making-requests-when-getting-updates
return SendMessage(
chat_id=message.chat.id, text="Hi from webhook!", reply_to_message_id=message.message_id
)
@dp.message_handler()
async def echo(message: types.Message):
# Regular request
# await bot.send_message(message.chat.id, message.text)
# or reply INTO webhook
return SendMessage(message.chat.id, message.text)
async def cmd_about(message: types.Message):
# In this function markdown utils are userd for formatting message text
return SendMessage(
message.chat.id,
text(
bold("Hi! I'm just a simple telegram bot."),
"",
text("I'm powered by", bold("Python", Version(*sys.version_info[:]))),
text(
"With",
link(text("aiogram", aiogram.VERSION), "https://github.com/aiogram/aiogram"),
),
sep="\n",
),
parse_mode=ParseMode.MARKDOWN,
)
async def on_startup(dp):
await bot.set_webhook(WEBHOOK_URL)
# insert code here to run it after start
async def cancel(message: types.Message):
# Get current state context
state = dp.current_state(chat=message.chat.id, user=message.from_user.id)
async def on_shutdown(dp):
logging.warning('Shutting down..')
# If current user in any state - cancel it.
if await state.get_state() is not None:
await state.set_state(state=None)
return SendMessage(message.chat.id, "Current action is canceled.")
# Otherwise do nothing
# insert code here to run it before shutdown
async def unknown(message: types.Message):
"""
Handler for unknown messages.
"""
return SendMessage(
message.chat.id,
f"I don't know what to do with content type `{message.content_type()}`. Sorry :c",
)
async def cmd_id(message: types.Message):
"""
Return info about user.
"""
if message.reply_to_message:
target = message.reply_to_message.from_user
chat = message.chat
elif message.forward_from and message.chat.type == ChatType.PRIVATE:
target = message.forward_from
chat = message.forward_from or message.chat
else:
target = message.from_user
chat = message.chat
result_msg = [hbold("Info about user:"), f"First name: {target.first_name}"]
if target.last_name:
result_msg.append(f"Last name: {target.last_name}")
if target.username:
result_msg.append(f"Username: {target.mention}")
result_msg.append(f"User ID: {target.id}")
result_msg.extend([hbold("Chat:"), f"Type: {chat.type}", f"Chat ID: {chat.id}"])
if chat.type != ChatType.PRIVATE:
result_msg.append(f"Title: {chat.title}")
else:
result_msg.append(f"Title: {chat.full_name}")
return SendMessage(
message.chat.id,
"\n".join(result_msg),
reply_to_message_id=message.message_id,
parse_mode=ParseMode.HTML,
)
async def on_startup(app):
# Demonstrate one of the available methods for registering handlers
# This command available only in main state (state=None)
dp.register_message_handler(cmd_start, commands=["start"])
# This handler is available in all states at any time.
dp.register_message_handler(cmd_about, commands=["help", "about"], state="*")
dp.register_message_handler(
unknown,
content_types=BAD_CONTENT,
func=lambda message: message.chat.type == ChatType.PRIVATE,
)
# You are able to register one function handler for multiple conditions
dp.register_message_handler(cancel, commands=["cancel"], state="*")
dp.register_message_handler(
cancel, func=lambda message: message.text.lower().strip() in ["cancel"], state="*"
)
dp.register_message_handler(cmd_id, commands=["id"], state="*")
dp.register_message_handler(
cmd_id,
func=lambda message: message.forward_from
or message.reply_to_message
and message.chat.type == ChatType.PRIVATE,
state="*",
)
# Get current webhook status
webhook = await bot.get_webhook_info()
# If URL is bad
if webhook.url != WEBHOOK_URL:
# If URL doesnt match current - remove webhook
if not webhook.url:
await bot.delete_webhook()
# Set new URL for webhook
await bot.set_webhook(WEBHOOK_URL, certificate=open(WEBHOOK_SSL_CERT, "rb"))
# If you want to use free certificate signed by LetsEncrypt you need to set only URL without sending certificate.
async def on_shutdown(app):
"""
Graceful shutdown. This method is recommended by aiohttp docs.
"""
# Remove webhook.
# Remove webhook (not acceptable in some cases)
await bot.delete_webhook()
# Close Redis connection.
# Close DB connection (if used)
await dp.storage.close()
await dp.storage.wait_closed()
logging.warning('Bye!')
if __name__ == "__main__":
# Get instance of :class:`aiohttp.web.Application` with configured router.
app = get_new_configured_app(dispatcher=dp, path=WEBHOOK_URL_PATH)
# Setup event handlers.
app.on_startup.append(on_startup)
app.on_shutdown.append(on_shutdown)
# Generate SSL context
context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
context.load_cert_chain(WEBHOOK_SSL_CERT, WEBHOOK_SSL_PRIV)
# Start web-application.
web.run_app(app, host=WEBAPP_HOST, port=WEBAPP_PORT, ssl_context=context)
# Note:
# If you start your bot using nginx or Apache web server, SSL context is not required.
# Otherwise you need to set `ssl_context` parameter.
if __name__ == '__main__':
start_webhook(
dispatcher=dp,
webhook_path=WEBHOOK_PATH,
on_startup=on_startup,
on_shutdown=on_shutdown,
skip_updates=True,
host=WEBAPP_HOST,
port=WEBAPP_PORT,
)

50
examples/webhook_example_2.py
View file

@ -1,50 +0,0 @@
import asyncio
import logging
from aiogram import Bot, types
from aiogram.dispatcher import Dispatcher
from aiogram.utils.executor import start_webhook
API_TOKEN = "BOT TOKEN HERE"
# webhook settings
WEBHOOK_HOST = "https://your.domain"
WEBHOOK_PATH = "/path/to/api"
WEBHOOK_URL = f"{WEBHOOK_HOST}{WEBHOOK_PATH}"
# webserver settings
WEBAPP_HOST = "localhost" # or ip
WEBAPP_PORT = 3001
logging.basicConfig(level=logging.INFO)
loop = asyncio.get_event_loop()
bot = Bot(token=API_TOKEN, loop=loop)
dp = Dispatcher(bot)
@dp.message_handler()
async def echo(message: types.Message):
await bot.send_message(message.chat.id, message.text)
async def on_startup(dp):
await bot.set_webhook(WEBHOOK_URL)
# insert code here to run it after start
async def on_shutdown(dp):
# insert code here to run it before shutdown
pass
if __name__ == "__main__":
start_webhook(
dispatcher=dp,
webhook_path=WEBHOOK_PATH,
on_startup=on_startup,
on_shutdown=on_shutdown,
skip_updates=True,
host=WEBAPP_HOST,
port=WEBAPP_PORT,
)

176
examples/webhook_example_old.py Normal file
View file

@ -0,0 +1,176 @@
"""
Example outdated
"""
import asyncio
import ssl
import sys
from aiohttp import web
import aiogram
from aiogram import Bot, types
from aiogram.contrib.fsm_storage.memory import MemoryStorage
from aiogram.dispatcher import Dispatcher
from aiogram.dispatcher.webhook import get_new_configured_app, SendMessage
from aiogram.types import ChatType, ParseMode, ContentTypes
from aiogram.utils.markdown import hbold, bold, text, link
TOKEN = 'BOT TOKEN HERE'
WEBHOOK_HOST = 'example.com' # Domain name or IP addres which your bot is located.
WEBHOOK_PORT = 443 # Telegram Bot API allows only for usage next ports: 443, 80, 88 or 8443
WEBHOOK_URL_PATH = '/webhook' # Part of URL
# This options needed if you use self-signed SSL certificate
# Instructions: https://core.telegram.org/bots/self-signed
WEBHOOK_SSL_CERT = './webhook_cert.pem' # Path to the ssl certificate
WEBHOOK_SSL_PRIV = './webhook_pkey.pem' # Path to the ssl private key
WEBHOOK_URL = f"https://{WEBHOOK_HOST}:{WEBHOOK_PORT}{WEBHOOK_URL_PATH}"
# Web app settings:
# Use LAN address to listen webhooks
# User any available port in range from 1024 to 49151 if you're using proxy, or WEBHOOK_PORT if you're using direct webhook handling
WEBAPP_HOST = 'localhost'
WEBAPP_PORT = 3001
BAD_CONTENT = ContentTypes.PHOTO & ContentTypes.DOCUMENT & ContentTypes.STICKER & ContentTypes.AUDIO
bot = Bot(TOKEN)
storage = MemoryStorage()
dp = Dispatcher(bot, storage=storage)
async def cmd_start(message: types.Message):
# Yep. aiogram allows to respond into webhook.
# https://core.telegram.org/bots/api#making-requests-when-getting-updates
return SendMessage(chat_id=message.chat.id, text='Hi from webhook!',
reply_to_message_id=message.message_id)
async def cmd_about(message: types.Message):
# In this function markdown utils are userd for formatting message text
return SendMessage(message.chat.id, text(
bold('Hi! I\'m just a simple telegram bot.'),
'',
text('I\'m powered by', bold('Python', Version(*sys.version_info[:]))),
text('With', link(text('aiogram', aiogram.VERSION), 'https://github.com/aiogram/aiogram')),
sep='\n'
), parse_mode=ParseMode.MARKDOWN)
async def cancel(message: types.Message):
# Get current state context
state = dp.current_state(chat=message.chat.id, user=message.from_user.id)
# If current user in any state - cancel it.
if await state.get_state() is not None:
await state.set_state(state=None)
return SendMessage(message.chat.id, 'Current action is canceled.')
# Otherwise do nothing
async def unknown(message: types.Message):
"""
Handler for unknown messages.
"""
return SendMessage(message.chat.id,
f"I don\'t know what to do with content type `{message.content_type()}`. Sorry :c")
async def cmd_id(message: types.Message):
"""
Return info about user.
"""
if message.reply_to_message:
target = message.reply_to_message.from_user
chat = message.chat
elif message.forward_from and message.chat.type == ChatType.PRIVATE:
target = message.forward_from
chat = message.forward_from or message.chat
else:
target = message.from_user
chat = message.chat
result_msg = [hbold('Info about user:'),
f"First name: {target.first_name}"]
if target.last_name:
result_msg.append(f"Last name: {target.last_name}")
if target.username:
result_msg.append(f"Username: {target.mention}")
result_msg.append(f"User ID: {target.id}")
result_msg.extend([hbold('Chat:'),
f"Type: {chat.type}",
f"Chat ID: {chat.id}"])
if chat.type != ChatType.PRIVATE:
result_msg.append(f"Title: {chat.title}")
else:
result_msg.append(f"Title: {chat.full_name}")
return SendMessage(message.chat.id, '\n'.join(result_msg), reply_to_message_id=message.message_id,
parse_mode=ParseMode.HTML)
async def on_startup(app):
# Demonstrate one of the available methods for registering handlers
# This command available only in main state (state=None)
dp.register_message_handler(cmd_start, commands=['start'])
# This handler is available in all states at any time.
dp.register_message_handler(cmd_about, commands=['help', 'about'], state='*')
dp.register_message_handler(unknown, content_types=BAD_CONTENT,
func=lambda message: message.chat.type == ChatType.PRIVATE)
# You are able to register one function handler for multiple conditions
dp.register_message_handler(cancel, commands=['cancel'], state='*')
dp.register_message_handler(cancel, func=lambda message: message.text.lower().strip() in ['cancel'], state='*')
dp.register_message_handler(cmd_id, commands=['id'], state='*')
dp.register_message_handler(cmd_id, func=lambda message: message.forward_from or
message.reply_to_message and
message.chat.type == ChatType.PRIVATE, state='*')
# Get current webhook status
webhook = await bot.get_webhook_info()
# If URL is bad
if webhook.url != WEBHOOK_URL:
# If URL doesnt match current - remove webhook
if not webhook.url:
await bot.delete_webhook()
# Set new URL for webhook
await bot.set_webhook(WEBHOOK_URL, certificate=open(WEBHOOK_SSL_CERT, 'rb'))
# If you want to use free certificate signed by LetsEncrypt you need to set only URL without sending certificate.
async def on_shutdown(app):
"""
Graceful shutdown. This method is recommended by aiohttp docs.
"""
# Remove webhook.
await bot.delete_webhook()
# Close Redis connection.
await dp.storage.close()
await dp.storage.wait_closed()
if __name__ == '__main__':
# Get instance of :class:`aiohttp.web.Application` with configured router.
app = get_new_configured_app(dispatcher=dp, path=WEBHOOK_URL_PATH)
# Setup event handlers.
app.on_startup.append(on_startup)
app.on_shutdown.append(on_shutdown)
# Generate SSL context
context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
context.load_cert_chain(WEBHOOK_SSL_CERT, WEBHOOK_SSL_PRIV)
# Start web-application.
web.run_app(app, host=WEBAPP_HOST, port=WEBAPP_PORT, ssl_context=context)
# Note:
# If you start your bot using nginx or Apache web server, SSL context is not required.
# Otherwise you need to set `ssl_context` parameter.