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 branch 'dev-1.x' into filters-factory

Browse source
This commit is contained in:
Alex Root Junior 2018-04-10 02:51:11 +03:00
commit fbcbf9ade4
27 changed files with 1448 additions and 169 deletions
Download patch file Download diff file Expand all files Collapse all files

26
.gitignore vendored
View file

@ -33,6 +33,7 @@ pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.pytest_cache/
.coverage
.coverage.*
.cache
@ -43,37 +44,14 @@ coverage.xml
# Sphinx documentation
docs/_build/
# pyenv
.python-version
# virtualenv
.venv
venv/
ENV/
### JetBrains template
# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion, Android Studio and Webstorm
# Reference: https://intellij-support.jetbrains.com/hc/en-us/articles/206544839
# JetBrains
.idea/
# User-specific stuff:
.idea/**/workspace.xml
.idea/**/tasks.xml
.idea/dictionaries
# Sensitive or high-churn files:
.idea/**/dataSources/
.idea/**/dataSources.ids
.idea/**/dataSources.xml
.idea/**/dataSources.local.xml
.idea/**/sqlDataSources.xml
.idea/**/dynamic.xml
.idea/**/uiDesigner.xml
## File-based project format:
*.iws
# Current project
experiment.py

2
Makefile
View file

@ -13,7 +13,7 @@ clean:
find . -name '*.pyo' -exec $(RM) {} +
find . -name '*~' -exec $(RM) {} +
find . -name '__pycache__' -exec $(RM) {} +
$(RM) build/ dist/ docs/build/ .tox/ .cache/ *.egg-info
$(RM) build/ dist/ docs/build/ .tox/ .cache/ .pytest_cache/ *.egg-info
tag:
@echo "Add tag: '$(AIOGRAM_VERSION)'"

2
aiogram/__init__.py
View file

@ -20,7 +20,7 @@ else:
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
VERSION = Version(1, 1, 1, stage=Stage.DEV, build=0)
VERSION = Version(1, 2, 3, stage=Stage.DEV, build=0)
API_VERSION = Version(3, 6)
__version__ = VERSION.version

46
aiogram/bot/api.py
View file

@ -1,12 +1,12 @@
import os
import logging
import os
from http import HTTPStatus
import aiohttp
from .. import types
from ..utils import json
from ..utils import exceptions
from ..utils import json
from ..utils.helper import Helper, HelperMode, Item
# Main aiogram logger
@ -67,10 +67,50 @@ async def _check_result(method_name, response):
elif 'migrate_to_chat_id' in result_json:
raise exceptions.MigrateToChat(result_json['migrate_to_chat_id'])
elif response.status == HTTPStatus.BAD_REQUEST:
if exceptions.MessageNotModified.check(description):
exceptions.MessageNotModified.throw()
elif exceptions.MessageToForwardNotFound.check(description):
exceptions.MessageToForwardNotFound.throw()
elif exceptions.MessageIdentifierNotSpecified.check(description):
exceptions.MessageIdentifierNotSpecified.throw()
elif exceptions.ChatNotFound.check(description):
exceptions.ChatNotFound.throw()
elif exceptions.InvalidQueryID.check(description):
exceptions.InvalidQueryID.throw()
elif exceptions.InvalidHTTPUrlContent.check(description):
exceptions.InvalidHTTPUrlContent.throw()
elif exceptions.GroupDeactivated.check(description):
exceptions.GroupDeactivated.throw()
elif exceptions.WrongFileIdentifier.check(description):
exceptions.WrongFileIdentifier.throw()
elif exceptions.InvalidPeerID.check(description):
exceptions.InvalidPeerID.throw()
elif exceptions.WebhookRequireHTTPS.check(description):
exceptions.WebhookRequireHTTPS.throw()
elif exceptions.BadWebhookPort.check(description):
exceptions.BadWebhookPort.throw()
elif exceptions.CantParseUrl.check(description):
exceptions.CantParseUrl.throw()
elif exceptions.PhotoAsInputFileRequired.check(description):
exceptions.PhotoAsInputFileRequired.throw()
raise exceptions.BadRequest(description)
elif response.status == HTTPStatus.NOT_FOUND:
if exceptions.MethodNotKnown.check(description):
exceptions.MethodNotKnown.throw()
raise exceptions.NotFound(description)
elif response.status == HTTPStatus.CONFLICT:
if exceptions.TerminatedByOtherGetUpdates.check(description):
exceptions.TerminatedByOtherGetUpdates.throw()
if exceptions.CantGetUpdates.check(description):
exceptions.CantGetUpdates.throw()
raise exceptions.ConflictError(description)
elif response.status in [HTTPStatus.UNAUTHORIZED, HTTPStatus.FORBIDDEN]:
if exceptions.BotKicked.check(description):
exceptions.BotKicked.throw()
elif exceptions.BotBlocked.check(description):
exceptions.BotBlocked.throw()
elif exceptions.UserDeactivated.check(description):
exceptions.UserDeactivated.throw()
raise exceptions.Unauthorized(description)
elif response.status == HTTPStatus.REQUEST_ENTITY_TOO_LARGE:
raise exceptions.NetworkError('File too large for uploading. '
@ -161,7 +201,7 @@ class Methods(Helper):
"""
Helper for Telegram API Methods listed on https://core.telegram.org/bots/api
List is updated to Bot API 3.5
List is updated to Bot API 3.6
"""
mode = HelperMode.lowerCamelCase

13
aiogram/bot/base.py
View file

@ -1,8 +1,10 @@
import asyncio
import io
import ssl
from typing import Dict, List, Optional, Union
import aiohttp
import certifi
from . import api
from ..types import ParseMode, base
@ -55,9 +57,11 @@ class BaseBot:
self.loop = loop
# aiohttp main session
self.session = aiohttp.ClientSession(
connector=aiohttp.TCPConnector(limit=connections_limit),
loop=self.loop, json_serialize=json.dumps)
ssl_context = ssl.create_default_context(cafile=certifi.where())
connector = aiohttp.TCPConnector(limit=connections_limit, ssl_context=ssl_context,
loop=self.loop)
self.session = aiohttp.ClientSession(connector=connector, loop=self.loop,
json_serialize=json.dumps)
# Temp sessions
self._temp_sessions = []
@ -68,7 +72,8 @@ class BaseBot:
self.parse_mode = parse_mode
def __del__(self):
asyncio.ensure_future(self.close())
# asyncio.ensure_future(self.close())
pass
async def close(self):
"""

72
aiogram/bot/bot.py
View file

@ -47,8 +47,8 @@ class Bot(BaseBot):
:return: destination
"""
file = await self.get_file(file_id)
return await self.download_file(file_path=file.file_path, destination=destination, timeout=timeout,
chunk_size=chunk_size, seek=seek)
return await self.download_file(file_path=file.file_path, destination=destination,
timeout=timeout, chunk_size=chunk_size, seek=seek)
# === Getting updates ===
# https://core.telegram.org/bots/api#getting-updates
@ -231,6 +231,7 @@ class Bot(BaseBot):
async def send_photo(self, chat_id: typing.Union[base.Integer, base.String],
photo: typing.Union[base.InputFile, base.String],
caption: typing.Union[base.String, None] = None,
parse_mode: typing.Union[base.String, None] = None,
disable_notification: typing.Union[base.Boolean, None] = None,
reply_to_message_id: typing.Union[base.Integer, None] = None,
reply_markup: typing.Union[types.InlineKeyboardMarkup,
@ -248,6 +249,9 @@ class Bot(BaseBot):
:type photo: :obj:`typing.Union[base.InputFile, base.String]`
:param caption: Photo caption (may also be used when resending photos by file_id), 0-200 characters
:type caption: :obj:`typing.Union[base.String, None]`
:param parse_mode: Send Markdown or HTML, if you want Telegram apps to show bold, italic,
fixed-width text or inline URLs in your bot's message.
:type parse_mode: :obj:`typing.Union[base.String, None]`
:param disable_notification: Sends the message silently. Users will receive a notification with no sound.
:type disable_notification: :obj:`typing.Union[base.Boolean, None]`
:param reply_to_message_id: If the message is a reply, ID of the original message
@ -260,6 +264,9 @@ class Bot(BaseBot):
"""
reply_markup = prepare_arg(reply_markup)
payload = generate_payload(**locals(), exclude=['photo'])
if self.parse_mode:
payload.setdefault('parse_mode', self.parse_mode)
result = await self.send_file('photo', api.Methods.SEND_PHOTO, photo, payload)
return types.Message(**result)
@ -267,6 +274,7 @@ class Bot(BaseBot):
async def send_audio(self, chat_id: typing.Union[base.Integer, base.String],
audio: typing.Union[base.InputFile, base.String],
caption: typing.Union[base.String, None] = None,
parse_mode: typing.Union[base.String, None] = None,
duration: typing.Union[base.Integer, None] = None,
performer: typing.Union[base.String, None] = None,
title: typing.Union[base.String, None] = None,
@ -290,6 +298,9 @@ class Bot(BaseBot):
:type audio: :obj:`typing.Union[base.InputFile, base.String]`
:param caption: Audio caption, 0-200 characters
:type caption: :obj:`typing.Union[base.String, None]`
:param parse_mode: Send Markdown or HTML, if you want Telegram apps to show bold, italic,
fixed-width text or inline URLs in your bot's message.
:type parse_mode: :obj:`typing.Union[base.String, None]`
:param duration: Duration of the audio in seconds
:type duration: :obj:`typing.Union[base.Integer, None]`
:param performer: Performer
@ -308,6 +319,9 @@ class Bot(BaseBot):
"""
reply_markup = prepare_arg(reply_markup)
payload = generate_payload(**locals(), exclude=['audio'])
if self.parse_mode:
payload.setdefault('parse_mode', self.parse_mode)
result = await self.send_file('audio', api.Methods.SEND_AUDIO, audio, payload)
return types.Message(**result)
@ -315,6 +329,7 @@ class Bot(BaseBot):
async def send_document(self, chat_id: typing.Union[base.Integer, base.String],
document: typing.Union[base.InputFile, base.String],
caption: typing.Union[base.String, None] = None,
parse_mode: typing.Union[base.String, None] = None,
disable_notification: typing.Union[base.Boolean, None] = None,
reply_to_message_id: typing.Union[base.Integer, None] = None,
reply_markup: typing.Union[types.InlineKeyboardMarkup,
@ -334,6 +349,9 @@ class Bot(BaseBot):
:type document: :obj:`typing.Union[base.InputFile, base.String]`
:param caption: Document caption (may also be used when resending documents by file_id), 0-200 characters
:type caption: :obj:`typing.Union[base.String, None]`
:param parse_mode: Send Markdown or HTML, if you want Telegram apps to show bold, italic,
fixed-width text or inline URLs in your bot's message.
:type parse_mode: :obj:`typing.Union[base.String, None]`
:param disable_notification: Sends the message silently. Users will receive a notification with no sound.
:type disable_notification: :obj:`typing.Union[base.Boolean, None]`
:param reply_to_message_id: If the message is a reply, ID of the original message
@ -346,6 +364,9 @@ class Bot(BaseBot):
"""
reply_markup = prepare_arg(reply_markup)
payload = generate_payload(**locals(), exclude=['document'])
if self.parse_mode:
payload.setdefault('parse_mode', self.parse_mode)
result = await self.send_file('document', api.Methods.SEND_DOCUMENT, document, payload)
return types.Message(**result)
@ -356,6 +377,7 @@ class Bot(BaseBot):
width: typing.Union[base.Integer, None] = None,
height: typing.Union[base.Integer, None] = None,
caption: typing.Union[base.String, None] = None,
parse_mode: typing.Union[base.String, None] = None,
supports_streaming: typing.Union[base.Boolean, None] = None,
disable_notification: typing.Union[base.Boolean, None] = None,
reply_to_message_id: typing.Union[base.Integer, None] = None,
@ -381,6 +403,9 @@ class Bot(BaseBot):
:type height: :obj:`typing.Union[base.Integer, None]`
:param caption: Video caption (may also be used when resending videos by file_id), 0-200 characters
:type caption: :obj:`typing.Union[base.String, None]`
:param parse_mode: Send Markdown or HTML, if you want Telegram apps to show bold, italic,
fixed-width text or inline URLs in your bot's message.
:type parse_mode: :obj:`typing.Union[base.String, None]`
:param supports_streaming: Pass True, if the uploaded video is suitable for streaming
:type supports_streaming: :obj:`typing.Union[base.Boolean, None]`
:param disable_notification: Sends the message silently. Users will receive a notification with no sound.
@ -395,6 +420,9 @@ class Bot(BaseBot):
"""
reply_markup = prepare_arg(reply_markup)
payload = generate_payload(**locals(), exclude=['video'])
if self.parse_mode:
payload.setdefault('parse_mode', self.parse_mode)
result = await self.send_file('video', api.Methods.SEND_VIDEO, video, payload)
return types.Message(**result)
@ -402,6 +430,7 @@ class Bot(BaseBot):
async def send_voice(self, chat_id: typing.Union[base.Integer, base.String],
voice: typing.Union[base.InputFile, base.String],
caption: typing.Union[base.String, None] = None,
parse_mode: typing.Union[base.String, None] = None,
duration: typing.Union[base.Integer, None] = None,
disable_notification: typing.Union[base.Boolean, None] = None,
reply_to_message_id: typing.Union[base.Integer, None] = None,
@ -424,6 +453,9 @@ class Bot(BaseBot):
:type voice: :obj:`typing.Union[base.InputFile, base.String]`
:param caption: Voice message caption, 0-200 characters
:type caption: :obj:`typing.Union[base.String, None]`
:param parse_mode: Send Markdown or HTML, if you want Telegram apps to show bold, italic,
fixed-width text or inline URLs in your bot's message.
:type parse_mode: :obj:`typing.Union[base.String, None]`
:param duration: Duration of the voice message in seconds
:type duration: :obj:`typing.Union[base.Integer, None]`
:param disable_notification: Sends the message silently. Users will receive a notification with no sound.
@ -438,6 +470,9 @@ class Bot(BaseBot):
"""
reply_markup = prepare_arg(reply_markup)
payload = generate_payload(**locals(), exclude=['voice'])
if self.parse_mode:
payload.setdefault('parse_mode', self.parse_mode)
result = await self.send_file('voice', api.Methods.SEND_VOICE, voice, payload)
return types.Message(**result)
@ -516,8 +551,9 @@ class Bot(BaseBot):
return [types.Message(**message) for message in result]
async def send_location(self, chat_id: typing.Union[base.Integer, base.String], latitude: base.Float,
longitude: base.Float, live_period: typing.Union[base.Integer, None] = None,
async def send_location(self, chat_id: typing.Union[base.Integer, base.String],
latitude: base.Float, longitude: base.Float,
live_period: typing.Union[base.Integer, None] = None,
disable_notification: typing.Union[base.Boolean, None] = None,
reply_to_message_id: typing.Union[base.Integer, None] = None,
reply_markup: typing.Union[types.InlineKeyboardMarkup,
@ -625,7 +661,8 @@ class Bot(BaseBot):
return types.Message(**result)
async def send_venue(self, chat_id: typing.Union[base.Integer, base.String],
latitude: base.Float, longitude: base.Float, title: base.String, address: base.String,
latitude: base.Float, longitude: base.Float,
title: base.String, address: base.String,
foursquare_id: typing.Union[base.String, None] = None,
disable_notification: typing.Union[base.Boolean, None] = None,
reply_to_message_id: typing.Union[base.Integer, None] = None,
@ -667,8 +704,8 @@ class Bot(BaseBot):
return types.Message(**result)
async def send_contact(self, chat_id: typing.Union[base.Integer, base.String],
phone_number: base.String,
first_name: base.String, last_name: typing.Union[base.String, None] = None,
phone_number: base.String, first_name: base.String,
last_name: typing.Union[base.String, None] = None,
disable_notification: typing.Union[base.Boolean, None] = None,
reply_to_message_id: typing.Union[base.Integer, None] = None,
reply_markup: typing.Union[types.InlineKeyboardMarkup,
@ -1182,7 +1219,8 @@ class Bot(BaseBot):
return result
async def answer_callback_query(self, callback_query_id: base.String, text: typing.Union[base.String, None] = None,
async def answer_callback_query(self, callback_query_id: base.String,
text: typing.Union[base.String, None] = None,
show_alert: typing.Union[base.Boolean, None] = None,
url: typing.Union[base.String, None] = None,
cache_time: typing.Union[base.Integer, None] = None) -> base.Boolean:
@ -1265,6 +1303,7 @@ class Bot(BaseBot):
message_id: typing.Union[base.Integer, None] = None,
inline_message_id: typing.Union[base.String, None] = None,
caption: typing.Union[base.String, None] = None,
parse_mode: typing.Union[base.String, None] = None,
reply_markup: typing.Union[types.InlineKeyboardMarkup,
None] = None) -> types.Message or base.Boolean:
"""
@ -1281,6 +1320,9 @@ class Bot(BaseBot):
:type inline_message_id: :obj:`typing.Union[base.String, None]`
:param caption: New caption of the message
:type caption: :obj:`typing.Union[base.String, None]`
:param parse_mode: Send Markdown or HTML, if you want Telegram apps to show bold, italic,
fixed-width text or inline URLs in your bot's message.
:type parse_mode: :obj:`typing.Union[base.String, None]`
:param reply_markup: A JSON-serialized object for an inline keyboard.
:type reply_markup: :obj:`typing.Union[types.InlineKeyboardMarkup, None]`
:return: On success, if edited message is sent by the bot, the edited Message is returned,
@ -1289,6 +1331,9 @@ class Bot(BaseBot):
"""
reply_markup = prepare_arg(reply_markup)
payload = generate_payload(**locals())
if self.parse_mode:
payload.setdefault('parse_mode', self.parse_mode)
result = await self.request(api.Methods.EDIT_MESSAGE_CAPTION, payload)
if isinstance(result, bool):
@ -1524,7 +1569,8 @@ class Bot(BaseBot):
return result
async def answer_inline_query(self, inline_query_id: base.String, results: typing.List[types.InlineQueryResult],
async def answer_inline_query(self, inline_query_id: base.String,
results: typing.List[types.InlineQueryResult],
cache_time: typing.Union[base.Integer, None] = None,
is_personal: typing.Union[base.Boolean, None] = None,
next_offset: typing.Union[base.String, None] = None,
@ -1570,8 +1616,9 @@ class Bot(BaseBot):
# === Payments ===
# https://core.telegram.org/bots/api#payments
async def send_invoice(self, chat_id: base.Integer, title: base.String, description: base.String,
payload: base.String, provider_token: base.String, start_parameter: base.String,
async def send_invoice(self, chat_id: base.Integer, title: base.String,
description: base.String, payload: base.String,
provider_token: base.String, start_parameter: base.String,
currency: base.String, prices: typing.List[types.LabeledPrice],
provider_data: typing.Union[typing.Dict, None] = None,
photo_url: typing.Union[base.String, None] = None,
@ -1783,7 +1830,8 @@ class Bot(BaseBot):
return types.Message(**result)
async def get_game_high_scores(self, user_id: base.Integer, chat_id: typing.Union[base.Integer, None] = None,
async def get_game_high_scores(self, user_id: base.Integer,
chat_id: typing.Union[base.Integer, None] = None,
message_id: typing.Union[base.Integer, None] = None,
inline_message_id: typing.Union[base.String,
None] = None) -> typing.List[types.GameHighScore]:

11
aiogram/contrib/fsm_storage/memory.py
View file

@ -86,19 +86,24 @@ class MemoryStorage(BaseStorage):
def has_bucket(self):
return True
async def get_bucket(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
async def get_bucket(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
default: typing.Optional[dict] = None) -> typing.Dict:
chat, user = self.check_address(chat=chat, user=user)
user = self._get_user(chat, user)
return user['bucket']
async def set_bucket(self, *, chat: typing.Union[str, int, None] = None, user: typing.Union[str, int, None] = None,
async def set_bucket(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
bucket: typing.Dict = None):
chat, user = self.check_address(chat=chat, user=user)
user = self._get_user(chat, user)
user['bucket'] = bucket
async def update_bucket(self, *, chat: typing.Union[str, int, None] = None,
async def update_bucket(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
bucket: typing.Dict = None, **kwargs):
chat, user = self.check_address(chat=chat, user=user)

11
aiogram/dispatcher/filters.py
View file

@ -4,7 +4,7 @@ import inspect
import re
import typing
from ..types import ContentType
from ..types import CallbackQuery, ContentType, Message
from ..utils import context
from ..utils.helper import Helper, HelperMode, Item
@ -246,9 +246,12 @@ class RegexpFilter(Filter):
self.regexp = re.compile(regexp, flags=re.IGNORECASE | re.MULTILINE)
super().__init__()
def check(self, message):
if message.text:
return bool(self.regexp.search(message.text))
def check(self, obj):
if isinstance(obj, Message) and obj.text:
return bool(self.regexp.search(obj.text))
elif isinstance(obj, CallbackQuery) and obj.data:
return bool(self.regexp.search(obj.data))
return False
class RegexpCommandsFilter(AsyncFilter):

2
aiogram/dispatcher/handler.py
View file

@ -72,7 +72,7 @@ class Handler:
context.set_value('handler', handler)
await self.dispatcher.middleware.trigger(f"process_{self.middleware_key}", args)
response = await handler(*args)
if results is not None:
if response is not None:
results.append(response)
if self.once:
break

109
aiogram/dispatcher/storage.py
View file

@ -1,5 +1,8 @@
import typing
from ..utils.deprecated import warn_deprecated as warn
from ..utils.exceptions import FSMStorageWarning
# Leak bucket
KEY = 'key'
LAST_CALL = 'called_at'
@ -12,13 +15,14 @@ THROTTLE_MANAGER = '$throttle_manager'
class BaseStorage:
"""
In states-storage you can save current user state and data for all steps
You are able to save current user's state
and data for all steps in states-storage
"""
async def close(self):
"""
Need override this method and use when application is shutdowns.
You can save data or etc.
You have to override this method and use when application shutdowns.
Perhaps you would like to save data and etc.
:return:
"""
@ -26,7 +30,7 @@ class BaseStorage:
async def wait_closed(self):
"""
You need override this method for all asynchronously storage's like Redis.
You have to override this method for all asynchronous storages (e.g., Redis).
:return:
"""
@ -37,34 +41,33 @@ class BaseStorage:
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None) -> (typing.Union[str, int], typing.Union[str, int]):
"""
In all methods of storage chat or user is always required.
If one of this is not presented, need set the missing value based on the presented.
In all storage's methods chat or user is always required.
If one of them is not provided, you have to set missing value based on the provided one.
This method performs the above action.
This method performs the check described above.
:param chat:
:param user:
:return:
"""
if chat is not None and user is not None:
return chat, user
elif user is None and chat is not None:
if chat is None and user is None:
raise ValueError('`user` or `chat` parameter is required but no one is provided!')
if user is None and chat is not None:
user = chat
return chat, user
elif user is not None and chat is None:
chat = user
return chat, user
raise ValueError('User or chat parameters is required but anyone is not presented!')
return chat, user
async def get_state(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
default: typing.Optional[str] = None) -> typing.Optional[str]:
"""
Get current state of user in chat. Return value stored in `default` parameter if record is not found.
Get current state of user in chat. Return `default` if no record is found.
Chat or user is always required. If one of this is not presented,
need set the missing value based on the presented
Chat or user is always required. If one of them is not provided,
you have to set missing value based on the provided one.
:param chat:
:param user:
@ -78,10 +81,10 @@ class BaseStorage:
user: typing.Union[str, int, None] = None,
default: typing.Optional[typing.Dict] = None) -> typing.Dict:
"""
Get state-data for user in chat. Return `default` if data is not presented in storage.
Get state-data for user in chat. Return `default` if no data is provided in storage.
Chat or user is always required. If one of this is not presented,
need set the missing value based on the presented
Chat or user is always required. If one of them is not provided,
you have to set missing value based on the provided one.
:param chat:
:param user:
@ -95,10 +98,10 @@ class BaseStorage:
user: typing.Union[str, int, None] = None,
state: typing.Optional[typing.AnyStr] = None):
"""
Setup new state for user in chat
Set new state for user in chat
Chat or user is always required. If one of this is not presented,
need set the missing value based on the presented
Chat or user is always required. If one of them is not provided,
you have to set missing value based on the provided one.
:param chat:
:param user:
@ -113,8 +116,8 @@ class BaseStorage:
"""
Set data for user in chat
Chat or user is always required. If one of this is not presented,
need set the missing value based on the presented
Chat or user is always required. If one of them is not provided,
you have to set missing value based on the provided one.
:param chat:
:param user:
@ -132,8 +135,8 @@ class BaseStorage:
You can use data parameter or|and kwargs.
Chat or user is always required. If one of this is not presented,
need set the missing value based on the presented
Chat or user is always required. If one of them is not provided,
you have to set missing value based on the provided one.
:param data:
:param chat:
@ -147,10 +150,10 @@ class BaseStorage:
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None):
"""
Reset data dor user in chat.
Reset data for user in chat.
Chat or user is always required. If one of this is not presented,
need set the missing value based on the presented
Chat or user is always required. If one of them is not provided,
you have to set missing value based on the provided one.
:param chat:
:param user:
@ -163,10 +166,11 @@ class BaseStorage:
user: typing.Union[str, int, None] = None,
with_data: typing.Optional[bool] = True):
"""
Reset state for user in chat. You can use this method for finish conversations.
Reset state for user in chat.
You may desire to use this method when finishing conversations.
Chat or user is always required. If one of this is not presented,
need set the missing value based on the presented
you have to set missing value based on the provided one.
:param chat:
:param user:
@ -184,8 +188,8 @@ class BaseStorage:
"""
Finish conversation for user in chat.
Chat or user is always required. If one of this is not presented,
need set the missing value based on the presented
Chat or user is always required. If one of them is not provided,
you have to set missing value based on the provided one.
:param chat:
:param user:
@ -201,10 +205,10 @@ class BaseStorage:
user: typing.Union[str, int, None] = None,
default: typing.Optional[dict] = None) -> typing.Dict:
"""
Get state-data for user in chat. Return `default` if data is not presented in storage.
Get bucket for user in chat. Return `default` if no data is provided in storage.
Chat or user is always required. If one of this is not presented,
need set the missing value based on the presented
Chat or user is always required. If one of them is not provided,
you have to set missing value based on the provided one.
:param chat:
:param user:
@ -218,10 +222,10 @@ class BaseStorage:
user: typing.Union[str, int, None] = None,
bucket: typing.Dict = None):
"""
Set data for user in chat
Set bucket for user in chat
Chat or user is always required. If one of this is not presented,
need set the missing value based on the presented
Chat or user is always required. If one of them is not provided,
you have to set missing value based on the provided one.
:param chat:
:param user:
@ -235,12 +239,12 @@ class BaseStorage:
bucket: typing.Dict = None,
**kwargs):
"""
Update data for user in chat
Update bucket for user in chat
You can use data parameter or|and kwargs.
You can use bucket parameter or|and kwargs.
Chat or user is always required. If one of this is not presented,
need set the missing value based on the presented
Chat or user is always required. If one of them is not provided,
you have to set missing value based on the provided one.
:param bucket:
:param chat:
@ -254,10 +258,10 @@ class BaseStorage:
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None):
"""
Reset data dor user in chat.
Reset bucket dor user in chat.
Chat or user is always required. If one of this is not presented,
need set the missing value based on the presented
Chat or user is always required. If one of them is not provided,
you have to set missing value based on the provided one.
:param chat:
:param user:
@ -323,22 +327,29 @@ class DisabledStorage(BaseStorage):
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
default: typing.Optional[str] = None) -> typing.Dict:
self._warn()
return {}
async def update_data(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
data: typing.Dict = None, **kwargs):
pass
self._warn()
async def set_state(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
state: typing.Optional[typing.AnyStr] = None):
pass
self._warn()
async def set_data(self, *,
chat: typing.Union[str, int, None] = None,
user: typing.Union[str, int, None] = None,
data: typing.Dict = None):
pass
self._warn()
@staticmethod
def _warn():
warn(f"You haven’t set any storage yet so no states and no data will be saved. \n"
f"You can connect MemoryStorage for debug purposes or non-essential data.",
FSMStorageWarning, 5)

16
aiogram/types/chat.py
View file

@ -521,7 +521,7 @@ class ChatActions(helper.Helper):
:param sleep: sleep timeout
:return:
"""
await cls._do(cls.UPLOAD_PHOTO, sleep)
await cls._do(cls.RECORD_VIDEO, sleep)
@classmethod
async def upload_video(cls, sleep=None):
@ -531,7 +531,7 @@ class ChatActions(helper.Helper):
:param sleep: sleep timeout
:return:
"""
await cls._do(cls.RECORD_VIDEO, sleep)
await cls._do(cls.UPLOAD_VIDEO, sleep)
@classmethod
async def record_audio(cls, sleep=None):
@ -541,7 +541,7 @@ class ChatActions(helper.Helper):
:param sleep: sleep timeout
:return:
"""
await cls._do(cls.UPLOAD_VIDEO, sleep)
await cls._do(cls.RECORD_AUDIO, sleep)
@classmethod
async def upload_audio(cls, sleep=None):
@ -551,7 +551,7 @@ class ChatActions(helper.Helper):
:param sleep: sleep timeout
:return:
"""
await cls._do(cls.RECORD_AUDIO, sleep)
await cls._do(cls.UPLOAD_AUDIO, sleep)
@classmethod
async def upload_document(cls, sleep=None):
@ -561,7 +561,7 @@ class ChatActions(helper.Helper):
:param sleep: sleep timeout
:return:
"""
await cls._do(cls.UPLOAD_AUDIO, sleep)
await cls._do(cls.UPLOAD_DOCUMENT, sleep)
@classmethod
async def find_location(cls, sleep=None):
@ -571,7 +571,7 @@ class ChatActions(helper.Helper):
:param sleep: sleep timeout
:return:
"""
await cls._do(cls.UPLOAD_DOCUMENT, sleep)
await cls._do(cls.FIND_LOCATION, sleep)
@classmethod
async def record_video_note(cls, sleep=None):
@ -581,7 +581,7 @@ class ChatActions(helper.Helper):
:param sleep: sleep timeout
:return:
"""
await cls._do(cls.FIND_LOCATION, sleep)
await cls._do(cls.RECORD_VIDEO_NOTE, sleep)
@classmethod
async def upload_video_note(cls, sleep=None):
@ -591,4 +591,4 @@ class ChatActions(helper.Helper):
:param sleep: sleep timeout
:return:
"""
await cls._do(cls.RECORD_VIDEO_NOTE, sleep)
await cls._do(cls.UPLOAD_VIDEO_NOTE, sleep)

12
aiogram/types/inline_keyboard.py
View file

@ -34,7 +34,8 @@ class InlineKeyboardMarkup(base.TelegramObject):
Add buttons
:param args:
:return:
:return: self
:rtype: :obj:`types.InlineKeyboardMarkup`
"""
row = []
for index, button in enumerate(args, start=1):
@ -44,13 +45,15 @@ class InlineKeyboardMarkup(base.TelegramObject):
row = []
if len(row) > 0:
self.inline_keyboard.append(row)
return self
def row(self, *args):
"""
Add row
:param args:
:return:
:return: self
:rtype: :obj:`types.InlineKeyboardMarkup`
"""
btn_array = []
for button in args:
@ -63,11 +66,14 @@ class InlineKeyboardMarkup(base.TelegramObject):
Insert button to last row
:param button:
:return: self
:rtype: :obj:`types.InlineKeyboardMarkup`
"""
if self.inline_keyboard and len(self.inline_keyboard[-1] < self.row_width):
if self.inline_keyboard and len(self.inline_keyboard[-1]) < self.row_width:
self.inline_keyboard[-1].append(button)
else:
self.add(button)
return self
class InlineKeyboardButton(base.TelegramObject):

83
aiogram/types/input_file.py
View file

@ -1,10 +1,15 @@
import io
import logging
import os
import time
import aiohttp
from . import base
from ..bot import api
CHUNK_SIZE = 65536
log = logging.getLogger('aiogram')
@ -76,6 +81,84 @@ class InputFile(base.TelegramObject):
"""
return self.file
@classmethod
async def from_url(cls, url, filename=None, chunk_size=CHUNK_SIZE):
"""
Download file from URL
Manually is not required action. You can send urls instead!
:param url: target URL
:param filename: optional. set custom file name
:param chunk_size:
:return: InputFile
"""
conf = {
'downloaded': True,
'url': url
}
# Let's do magic with the filename
if filename:
filename_prefix, _, ext = filename.rpartition('.')
file_suffix = '.' + ext if ext else ''
else:
filename_prefix, _, ext = url.rpartition('/')[-1].rpartition('.')
file_suffix = '.' + ext if ext else ''
filename = filename_prefix + file_suffix
async with aiohttp.ClientSession() as session:
start = time.time()
async with session.get(url) as response:
# Save file in memory
file = await cls._process_stream(response, io.BytesIO(), chunk_size=chunk_size)
log.debug(f"File successful downloaded at {round(time.time() - start, 2)} seconds from '{url}'")
return cls(file, filename, conf=conf)
def save(self, filename, chunk_size=CHUNK_SIZE):
"""
Write file to disk
:param filename:
:param chunk_size:
"""
with open(filename, 'wb') as fp:
while True:
# Chunk writer
data = self.file.read(chunk_size)
if not data:
break
fp.write(data)
# Flush all data
fp.flush()
# Go to start of file.
if self.file.seekable():
self.file.seek(0)
@classmethod
async def _process_stream(cls, response, writer, chunk_size=CHUNK_SIZE):
"""
Transfer data
:param response:
:param writer:
:param chunk_size:
:return:
"""
while True:
chunk = await response.content.read(chunk_size)
if not chunk:
break
writer.write(chunk)
if writer.seekable():
writer.seek(0)
return writer
def to_python(self):
raise TypeError('Object of this type is not exportable!')

37
aiogram/types/message.py
View file

@ -95,10 +95,10 @@ class Message(base.TelegramObject):
return ContentType.VOICE[0]
if self.contact:
return ContentType.CONTACT[0]
if self.location:
return ContentType.LOCATION[0]
if self.venue:
return ContentType.VENUE[0]
if self.location:
return ContentType.LOCATION[0]
if self.new_chat_members:
return ContentType.NEW_CHAT_MEMBERS[0]
if self.left_chat_member:
@ -130,7 +130,7 @@ class Message(base.TelegramObject):
command, _, args = self.text.partition(' ')
return command, args
def get_command(self):
def get_command(self, pure=False):
"""
Get command from message
@ -138,7 +138,10 @@ class Message(base.TelegramObject):
"""
command = self.get_full_command()
if command:
return command[0]
command = command[0]
if pure:
command, _, _ = command[1:].partition('@')
return command
def get_args(self):
"""
@ -181,7 +184,7 @@ class Message(base.TelegramObject):
return text
async def reply(self, text, parse_mode=None, disable_web_page_preview=None,
disable_notification=None, reply_markup=None, reply=False) -> 'Message':
disable_notification=None, reply_markup=None, reply=True) -> 'Message':
"""
Reply to this message
@ -630,6 +633,30 @@ class Message(base.TelegramObject):
"""
return await self.bot.delete_message(self.chat.id, self.message_id)
async def reply_sticker(self, sticker: typing.Union[base.InputFile, base.String],
disable_notification: typing.Union[base.Boolean, None] = None,
reply_markup=None, reply=True) -> 'Message':
"""
Use this method to send .webp stickers.
Source: https://core.telegram.org/bots/api#sendsticker
:param sticker: Sticker to send.
:type sticker: :obj:`typing.Union[base.InputFile, base.String]`
:param disable_notification: Sends the message silently. Users will receive a notification with no sound.
:type disable_notification: :obj:`typing.Union[base.Boolean, None]`
:param reply_markup: Additional interface options.
:type reply_markup: :obj:`typing.Union[types.InlineKeyboardMarkup,
types.ReplyKeyboardMarkup, types.ReplyKeyboardRemove, types.ForceReply, None]`
:param reply: fill 'reply_to_message_id'
:return: On success, the sent Message is returned.
:rtype: :obj:`types.Message`
"""
return await self.bot.send_sticker(chat_id=self.chat.id, sticker=sticker,
disable_notification=disable_notification,
reply_to_message_id=self.message_id if reply else None,
reply_markup=reply_markup)
async def pin(self, disable_notification: bool = False):
"""
Pin message

19
aiogram/types/reply_keyboard.py
View file

@ -37,23 +37,26 @@ class ReplyKeyboardMarkup(base.TelegramObject):
Add buttons
:param args:
:return:
:return: self
:rtype: :obj:`types.ReplyKeyboardMarkup`
"""
row = []
for index, button in enumerate(args):
for index, button in enumerate(args, start=1):
row.append(button)
if index % self.row_width == 0:
self.keyboard.append(row)
row = []
if len(row) > 0:
self.keyboard.append(row)
return self
def row(self, *args):
"""
Add row
:param args:
:return:
:return: self
:rtype: :obj:`types.ReplyKeyboardMarkup`
"""
btn_array = []
for button in args:
@ -66,11 +69,14 @@ class ReplyKeyboardMarkup(base.TelegramObject):
Insert button to last row
:param button:
:return: self
:rtype: :obj:`types.ReplyKeyboardMarkup`
"""
if self.keyboard and len(self.keyboard[-1]) < self.row_width:
self.keyboard[-1].append(button)
else:
self.add(button)
return self
class KeyboardButton(base.TelegramObject):
@ -84,8 +90,11 @@ class KeyboardButton(base.TelegramObject):
request_contact: base.Boolean = fields.Field()
request_location: base.Boolean = fields.Field()
def __init__(self, text: base.String, request_contact: base.Boolean = None, request_location: base.Boolean = None):
super(KeyboardButton, self).__init__(text=text, request_contact=request_contact,
def __init__(self, text: base.String,
request_contact: base.Boolean = None,
request_location: base.Boolean = None):
super(KeyboardButton, self).__init__(text=text,
request_contact=request_contact,
request_location=request_location)

5
aiogram/types/user.py
View file

@ -36,7 +36,8 @@ class User(base.TelegramObject):
@property
def mention(self):
"""
You can get mention to user (If user have username, otherwise return full name)
You can get user's username to mention him
Full name will be returned if user has no username
:return: str
"""
@ -47,7 +48,7 @@ class User(base.TelegramObject):
@property
def locale(self) -> 'babel.core.Locale' or None:
"""
This property require `Babel <https://pypi.python.org/pypi/Babel>`_ module
This property requires `Babel <https://pypi.python.org/pypi/Babel>`_ module
:return: :class:`babel.core.Locale`
:raise: ImportError: when babel is not installed.

4
aiogram/utils/deprecated.py
View file

@ -69,7 +69,7 @@ def deprecated(reason):
raise TypeError(repr(type(reason)))
def warn_deprecated(message, warning=DeprecationWarning):
def warn_deprecated(message, warning=DeprecationWarning, stacklevel=2):
warnings.simplefilter('always', warning)
warnings.warn(message, category=warning, stacklevel=2)
warnings.warn(message, category=warning, stacklevel=stacklevel)
warnings.simplefilter('default', warning)

170
aiogram/utils/exceptions.py
View file

@ -1,6 +1,43 @@
"""
TelegramAPIError
ValidationError
Throttled
BadRequest
MessageError
MessageNotModified
MessageToForwardNotFound
MessageToDeleteNotFound
MessageIdentifierNotSpecified
ChatNotFound
InvalidQueryID
InvalidPeerID
InvalidHTTPUrlContent
WrongFileIdentifier
GroupDeactivated
BadWebhook
WebhookRequireHTTPS
BadWebhookPort
CantParseUrl
NotFound
MethodNotKnown
PhotoAsInputFileRequired
ConflictError
TerminatedByOtherGetUpdates
CantGetUpdates
Unauthorized
BotKicked
BotBlocked
UserDeactivated
NetworkError
RetryAfter
MigrateToChat
AIOGramWarning
TimeoutWarning
"""
import time
_PREFIXES = ['Error: ', '[Error]: ', 'Bad Request: ', 'Conflict: ']
_PREFIXES = ['Error: ', '[Error]: ', 'Bad Request: ', 'Conflict: ', 'Not Found: ']
def _clean_message(text):
@ -11,10 +48,23 @@ def _clean_message(text):
class TelegramAPIError(Exception):
def __init__(self, message):
def __init__(self, message=None):
super(TelegramAPIError, self).__init__(_clean_message(message))
class _MatchErrorMixin:
match = ''
text = None
@classmethod
def check(cls, message):
return cls.match in message
@classmethod
def throw(cls):
raise cls(cls.text or cls.match)
class AIOGramWarning(Warning):
pass
@ -23,6 +73,10 @@ class TimeoutWarning(AIOGramWarning):
pass
class FSMStorageWarning(AIOGramWarning):
pass
class ValidationError(TelegramAPIError):
pass
@ -31,14 +85,124 @@ class BadRequest(TelegramAPIError):
pass
class MessageError(BadRequest):
pass
class MessageNotModified(MessageError, _MatchErrorMixin):
"""
Will be raised when you try to set new text is equals to current text.
"""
match = 'message is not modified'
class MessageToForwardNotFound(MessageError, _MatchErrorMixin):
"""
Will be raised when you try to forward very old or deleted or unknown message.
"""
match = 'message to forward not found'
class MessageToDeleteNotFound(MessageError, _MatchErrorMixin):
"""
Will be raised when you try to delete very old or deleted or unknown message.
"""
match = 'message to delete not found'
class MessageIdentifierNotSpecified(MessageError, _MatchErrorMixin):
match = 'message identifier is not specified'
class ChatNotFound(BadRequest, _MatchErrorMixin):
match = 'chat not found'
class InvalidQueryID(BadRequest, _MatchErrorMixin):
match = 'QUERY_ID_INVALID'
text = 'Invalid query ID'
class InvalidPeerID(BadRequest, _MatchErrorMixin):
match = 'PEER_ID_INVALID'
text = 'Invalid peer ID'
class InvalidHTTPUrlContent(BadRequest, _MatchErrorMixin):
match = 'Failed to get HTTP URL content'
class WrongFileIdentifier(BadRequest, _MatchErrorMixin):
match = 'wrong file identifier/HTTP URL specified'
class GroupDeactivated(BadRequest, _MatchErrorMixin):
match = 'group is deactivated'
class PhotoAsInputFileRequired(BadRequest, _MatchErrorMixin):
"""
Will be raised when you try to set chat photo from file ID.
"""
match = 'Photo should be uploaded as an InputFile'
class BadWebhook(BadRequest):
pass
class WebhookRequireHTTPS(BadRequest, _MatchErrorMixin):
match = 'HTTPS url must be provided for webhook'
text = 'bad webhook: ' + match
class BadWebhookPort(BadRequest, _MatchErrorMixin):
match = 'Webhook can be set up only on ports 80, 88, 443 or 8443'
text = 'bad webhook: ' + match
class CantParseUrl(BadRequest, _MatchErrorMixin):
match = 'can\'t parse URL'
class NotFound(TelegramAPIError):
pass
class MethodNotKnown(NotFound, _MatchErrorMixin):
match = 'method not found'
class ConflictError(TelegramAPIError):
pass
class TerminatedByOtherGetUpdates(ConflictError, _MatchErrorMixin):
match = 'terminated by other getUpdates request'
text = 'Terminated by other getUpdates request; ' \
'Make sure that only one bot instance is running'
class CantGetUpdates(ConflictError, _MatchErrorMixin):
match = 'can\'t use getUpdates method while webhook is active'
class Unauthorized(TelegramAPIError):
pass
class BotKicked(Unauthorized, _MatchErrorMixin):
match = 'Bot was kicked from a chat'
class BotBlocked(Unauthorized, _MatchErrorMixin):
match = 'bot was blocked by the user'
class UserDeactivated(Unauthorized, _MatchErrorMixin):
match = 'user is deactivated'
class NetworkError(TelegramAPIError):
pass
@ -55,7 +219,7 @@ class MigrateToChat(TelegramAPIError):
self.migrate_to_chat_id = chat_id
class Throttled(Exception):
class Throttled(TelegramAPIError):
def __init__(self, **kwargs):
from ..dispatcher.storage import DELTA, EXCEEDED_COUNT, KEY, LAST_CALL, RATE_LIMIT, RESULT
self.key = kwargs.pop(KEY, '<None>')

10
aiogram/utils/executor.py
View file

@ -11,15 +11,15 @@ async def _startup(dispatcher: Dispatcher, skip_updates=False, callback=None):
user = await dispatcher.bot.me
log.info(f"Bot: {user.full_name} [@{user.username}]")
if callable(callback):
await callback(dispatcher)
if skip_updates:
await dispatcher.reset_webhook(True)
count = await dispatcher.skip_updates()
if count:
log.warning(f"Skipped {count} updates.")
if callable(callback):
await callback(dispatcher)
async def _wh_startup(app):
callback = app.get('_startup_callback', None)
@ -39,6 +39,8 @@ async def _shutdown(dispatcher: Dispatcher, callback=None):
await dispatcher.storage.close()
await dispatcher.storage.wait_closed()
await dispatcher.bot.close()
async def _wh_shutdown(app):
callback = app.get('_shutdown_callback', None)
@ -87,5 +89,5 @@ def start_webhook(dispatcher, webhook_path, *, loop=None, skip_updates=None,
app.on_startup.append(_wh_startup)
app.on_shutdown.append(_wh_shutdown)
web.run_app(app, loop=loop, **kwargs)
web.run_app(app, **kwargs)
return app

11
aiogram/utils/markdown.py
View file

@ -185,3 +185,14 @@ def escape_md(*content, sep=' '):
:return:
"""
return _escape(_join(*content, sep=sep))
def hide_link(url):
"""
Hide URL (HTML only)
Can be used for adding an image to a text message
:param url:
:return:
"""
return f'<a href="{url}">&#8203;</a>'

2
dev_requirements.txt
View file

@ -10,3 +10,5 @@ wheel>=0.30.0
rethinkdb>=2.3.0
sphinx>=1.6.6
sphinx-rtd-theme>=0.2.4
aresponses
tox

2
examples/adwanced_executor_example.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/env python3
"""
II this example used ArgumentParser for configuring Your bot.
In this example used ArgumentParser for configuring Your bot.
Provided to start bot with webhook:
python adwanced_executor_example.py \

8
examples/webhook_example.py
View file

@ -25,6 +25,12 @@ 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 = ContentType.PHOTO & ContentType.DOCUMENT & ContentType.STICKER & ContentType.AUDIO
loop = asyncio.get_event_loop()
@ -160,7 +166,7 @@ if __name__ == '__main__':
context.load_cert_chain(WEBHOOK_SSL_CERT, WEBHOOK_SSL_PRIV)
# Start web-application.
web.run_app(app, host=WEBHOOK_HOST, port=WEBHOOK_PORT, ssl_context=context)
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.

1
requirements.txt
View file

@ -1,2 +1,3 @@
aiohttp>=2.3.5
Babel>=2.5.1
certifi>=2018.01.18

4
setup.py
View file

@ -1,11 +1,10 @@
#!/usr/bin/env python3
import sys
from distutils.core import setup
from warnings import warn
from pip.req import parse_requirements
from setuptools import PackageFinder
from setuptools import PackageFinder, setup
from aiogram import Stage, VERSION
@ -47,6 +46,7 @@ setup(
name='aiogram',
version=VERSION.version,
packages=PackageFinder.find(exclude=('tests', 'tests.*', 'examples.*', 'docs',)),
requires_python='>=3.6',
url='https://github.com/aiogram/aiogram',
license='MIT',
author='Alex Root Junior',

528
tests/test_bot.py
View file

@ -1,4 +1,526 @@
import aiogram
import aresponses
import pytest
# bot = aiogram.Bot('123456789:AABBCCDDEEFFaabbccddeeff-1234567890')
# TODO: mock for aiogram.bot.api.request and then test all AI methods.
from aiogram import Bot, types
TOKEN = '123456789:AABBCCDDEEFFaabbccddeeff-1234567890'
class FakeTelegram(aresponses.ResponsesMockServer):
def __init__(self, message_dict, **kwargs):
super().__init__(**kwargs)
self._body, self._headers = self.parse_data(message_dict)
async def __aenter__(self):
await super().__aenter__()
_response = self.Response(text=self._body, headers=self._headers, status=200, reason='OK')
self.add(self.ANY, response=_response)
@staticmethod
def parse_data(message_dict):
import json
_body = '{"ok":true,"result":' + json.dumps(message_dict) + '}'
_headers = {'Server': 'nginx/1.12.2',
'Date': 'Tue, 03 Apr 2018 16:59:54 GMT',
'Content-Type': 'application/json',
'Content-Length': str(len(_body)),
'Connection': 'keep-alive',
'Access-Control-Allow-Origin': '*',
'Access-Control-Allow-Methods': 'GET, POST, OPTIONS',
'Access-Control-Expose-Headers': 'Content-Length,Content-Type,Date,Server,Connection',
'Strict-Transport-Security': 'max-age=31536000; includeSubdomains'}
return _body, _headers
@pytest.yield_fixture()
@pytest.mark.asyncio
async def bot(event_loop):
""" Bot fixture """
_bot = Bot(TOKEN, loop=event_loop, parse_mode=types.ParseMode.MARKDOWN)
yield _bot
await _bot.close()
@pytest.mark.asyncio
async def test_get_me(bot: Bot, event_loop):
""" getMe method test """
from .types.dataset import USER
user = types.User(**USER)
async with FakeTelegram(message_dict=USER, loop=event_loop):
result = await bot.me
assert result == user
@pytest.mark.asyncio
async def test_send_message(bot: Bot, event_loop):
""" sendMessage method test """
from .types.dataset import MESSAGE
msg = types.Message(**MESSAGE)
async with FakeTelegram(message_dict=MESSAGE, loop=event_loop):
result = await bot.send_message(chat_id=msg.chat.id, text=msg.text)
assert result == msg
@pytest.mark.asyncio
async def test_forward_message(bot: Bot, event_loop):
""" forwardMessage method test """
from .types.dataset import FORWARDED_MESSAGE
msg = types.Message(**FORWARDED_MESSAGE)
async with FakeTelegram(message_dict=FORWARDED_MESSAGE, loop=event_loop):
result = await bot.forward_message(chat_id=msg.chat.id, from_chat_id=msg.forward_from_chat.id,
message_id=msg.forward_from_message_id)
assert result == msg
@pytest.mark.asyncio
async def test_send_photo(bot: Bot, event_loop):
""" sendPhoto method test with file_id """
from .types.dataset import MESSAGE_WITH_PHOTO, PHOTO
msg = types.Message(**MESSAGE_WITH_PHOTO)
photo = types.PhotoSize(**PHOTO)
async with FakeTelegram(message_dict=MESSAGE_WITH_PHOTO, loop=event_loop):
result = await bot.send_photo(msg.chat.id, photo=photo.file_id, caption=msg.caption,
parse_mode=types.ParseMode.HTML, disable_notification=False)
assert result == msg
@pytest.mark.asyncio
async def test_send_audio(bot: Bot, event_loop):
""" sendAudio method test with file_id """
from .types.dataset import MESSAGE_WITH_AUDIO
msg = types.Message(**MESSAGE_WITH_AUDIO)
async with FakeTelegram(message_dict=MESSAGE_WITH_AUDIO, loop=event_loop):
result = await bot.send_audio(chat_id=msg.chat.id, audio=msg.audio.file_id, caption=msg.caption,
parse_mode=types.ParseMode.HTML, duration=msg.audio.duration,
performer=msg.audio.performer, title=msg.audio.title, disable_notification=False)
assert result == msg
@pytest.mark.asyncio
async def test_send_document(bot: Bot, event_loop):
""" sendDocument method test with file_id """
from .types.dataset import MESSAGE_WITH_DOCUMENT
msg = types.Message(**MESSAGE_WITH_DOCUMENT)
async with FakeTelegram(message_dict=MESSAGE_WITH_DOCUMENT, loop=event_loop):
result = await bot.send_document(chat_id=msg.chat.id, document=msg.document.file_id, caption=msg.caption,
parse_mode=types.ParseMode.HTML, disable_notification=False)
assert result == msg
@pytest.mark.asyncio
async def test_send_video(bot: Bot, event_loop):
""" sendVideo method test with file_id """
from .types.dataset import MESSAGE_WITH_VIDEO, VIDEO
msg = types.Message(**MESSAGE_WITH_VIDEO)
video = types.Video(**VIDEO)
async with FakeTelegram(message_dict=MESSAGE_WITH_VIDEO, loop=event_loop):
result = await bot.send_video(chat_id=msg.chat.id, video=video.file_id, duration=video.duration,
width=video.width, height=video.height, caption=msg.caption,
parse_mode=types.ParseMode.HTML, supports_streaming=True,
disable_notification=False)
assert result == msg
@pytest.mark.asyncio
async def test_send_voice(bot: Bot, event_loop):
""" sendVoice method test with file_id """
from .types.dataset import MESSAGE_WITH_VOICE, VOICE
msg = types.Message(**MESSAGE_WITH_VOICE)
voice = types.Voice(**VOICE)
async with FakeTelegram(message_dict=MESSAGE_WITH_VOICE, loop=event_loop):
result = await bot.send_voice(chat_id=msg.chat.id, voice=voice.file_id, caption=msg.caption,
parse_mode=types.ParseMode.HTML, duration=voice.duration,
disable_notification=False)
assert result == msg
@pytest.mark.asyncio
async def test_send_video_note(bot: Bot, event_loop):
""" sendVideoNote method test with file_id """
from .types.dataset import MESSAGE_WITH_VIDEO_NOTE, VIDEO_NOTE
msg = types.Message(**MESSAGE_WITH_VIDEO_NOTE)
video_note = types.VideoNote(**VIDEO_NOTE)
async with FakeTelegram(message_dict=MESSAGE_WITH_VIDEO_NOTE, loop=event_loop):
result = await bot.send_video_note(chat_id=msg.chat.id, video_note=video_note.file_id,
duration=video_note.duration, length=video_note.length,
disable_notification=False)
assert result == msg
@pytest.mark.asyncio
async def test_send_media_group(bot: Bot, event_loop):
""" sendMediaGroup method test with file_id """
from .types.dataset import MESSAGE_WITH_MEDIA_GROUP, PHOTO
msg = types.Message(**MESSAGE_WITH_MEDIA_GROUP)
photo = types.PhotoSize(**PHOTO)
media = [types.InputMediaPhoto(media=photo.file_id), types.InputMediaPhoto(media=photo.file_id)]
async with FakeTelegram(message_dict=[MESSAGE_WITH_MEDIA_GROUP, MESSAGE_WITH_MEDIA_GROUP], loop=event_loop):
result = await bot.send_media_group(msg.chat.id, media=media, disable_notification=False)
assert len(result) == len(media)
assert result.pop().media_group_id
@pytest.mark.asyncio
async def test_send_location(bot: Bot, event_loop):
""" sendLocation method test """
from .types.dataset import MESSAGE_WITH_LOCATION, LOCATION
msg = types.Message(**MESSAGE_WITH_LOCATION)
location = types.Location(**LOCATION)
async with FakeTelegram(message_dict=MESSAGE_WITH_LOCATION, loop=event_loop):
result = await bot.send_location(msg.chat.id, latitude=location.latitude, longitude=location.longitude,
live_period=10, disable_notification=False)
assert result == msg
@pytest.mark.asyncio
async def test_edit_message_live_location(bot: Bot, event_loop):
""" editMessageLiveLocation method test """
from .types.dataset import MESSAGE_WITH_LOCATION, LOCATION
msg = types.Message(**MESSAGE_WITH_LOCATION)
location = types.Location(**LOCATION)
# editing bot message
async with FakeTelegram(message_dict=MESSAGE_WITH_LOCATION, loop=event_loop):
result = await bot.edit_message_live_location(chat_id=msg.chat.id, message_id=msg.message_id,
latitude=location.latitude, longitude=location.longitude)
assert result == msg
# editing user's message
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.edit_message_live_location(chat_id=msg.chat.id, message_id=msg.message_id,
latitude=location.latitude, longitude=location.longitude)
assert isinstance(result, bool) and result is True
@pytest.mark.asyncio
async def test_stop_message_live_location(bot: Bot, event_loop):
""" stopMessageLiveLocation method test """
from .types.dataset import MESSAGE_WITH_LOCATION
msg = types.Message(**MESSAGE_WITH_LOCATION)
# stopping bot message
async with FakeTelegram(message_dict=MESSAGE_WITH_LOCATION, loop=event_loop):
result = await bot.stop_message_live_location(chat_id=msg.chat.id, message_id=msg.message_id)
assert result == msg
# stopping user's message
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.stop_message_live_location(chat_id=msg.chat.id, message_id=msg.message_id)
assert isinstance(result, bool)
assert result is True
@pytest.mark.asyncio
async def test_send_venue(bot: Bot, event_loop):
""" sendVenue method test """
from .types.dataset import MESSAGE_WITH_VENUE, VENUE, LOCATION
msg = types.Message(**MESSAGE_WITH_VENUE)
location = types.Location(**LOCATION)
venue = types.Venue(**VENUE)
async with FakeTelegram(message_dict=MESSAGE_WITH_VENUE, loop=event_loop):
result = await bot.send_venue(msg.chat.id, latitude=location.latitude, longitude=location.longitude,
title=venue.title, address=venue.address, foursquare_id=venue.foursquare_id,
disable_notification=False)
assert result == msg
@pytest.mark.asyncio
async def test_send_contact(bot: Bot, event_loop):
""" sendContact method test """
from .types.dataset import MESSAGE_WITH_CONTACT, CONTACT
msg = types.Message(**MESSAGE_WITH_CONTACT)
contact = types.Contact(**CONTACT)
async with FakeTelegram(message_dict=MESSAGE_WITH_CONTACT, loop=event_loop):
result = await bot.send_contact(msg.chat.id, phone_number=contact.phone_number, first_name=contact.first_name,
last_name=contact.last_name, disable_notification=False)
assert result == msg
@pytest.mark.asyncio
async def test_send_chat_action(bot: Bot, event_loop):
""" sendChatAction method test """
from .types.dataset import CHAT
chat = types.Chat(**CHAT)
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.send_chat_action(chat_id=chat.id, action=types.ChatActions.TYPING)
assert isinstance(result, bool)
assert result is True
@pytest.mark.asyncio
async def test_get_user_profile_photo(bot: Bot, event_loop):
""" getUserProfilePhotos method test """
from .types.dataset import USER_PROFILE_PHOTOS, USER
user = types.User(**USER)
async with FakeTelegram(message_dict=USER_PROFILE_PHOTOS, loop=event_loop):
result = await bot.get_user_profile_photos(user_id=user.id, offset=1, limit=1)
assert isinstance(result, types.UserProfilePhotos)
@pytest.mark.asyncio
async def test_get_file(bot: Bot, event_loop):
""" getFile method test """
from .types.dataset import FILE
file = types.File(**FILE)
async with FakeTelegram(message_dict=FILE, loop=event_loop):
result = await bot.get_file(file_id=file.file_id)
assert isinstance(result, types.File)
@pytest.mark.asyncio
async def test_kick_chat_member(bot: Bot, event_loop):
""" kickChatMember method test """
from .types.dataset import USER, CHAT
user = types.User(**USER)
chat = types.Chat(**CHAT)
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.kick_chat_member(chat_id=chat.id, user_id=user.id, until_date=123)
assert isinstance(result, bool)
assert result is True
@pytest.mark.asyncio
async def test_unban_chat_member(bot: Bot, event_loop):
""" unbanChatMember method test """
from .types.dataset import USER, CHAT
user = types.User(**USER)
chat = types.Chat(**CHAT)
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.unban_chat_member(chat_id=chat.id, user_id=user.id)
assert isinstance(result, bool)
assert result is True
@pytest.mark.asyncio
async def test_restrict_chat_member(bot: Bot, event_loop):
""" restrictChatMember method test """
from .types.dataset import USER, CHAT
user = types.User(**USER)
chat = types.Chat(**CHAT)
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.restrict_chat_member(chat_id=chat.id, user_id=user.id, can_add_web_page_previews=False,
can_send_media_messages=False, can_send_messages=False,
can_send_other_messages=False, until_date=123)
assert isinstance(result, bool)
assert result is True
@pytest.mark.asyncio
async def test_promote_chat_member(bot: Bot, event_loop):
""" promoteChatMember method test """
from .types.dataset import USER, CHAT
user = types.User(**USER)
chat = types.Chat(**CHAT)
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.promote_chat_member(chat_id=chat.id, user_id=user.id, can_change_info=True,
can_delete_messages=True, can_edit_messages=True,
can_invite_users=True, can_pin_messages=True, can_post_messages=True,
can_promote_members=True, can_restrict_members=True)
assert isinstance(result, bool)
assert result is True
@pytest.mark.asyncio
async def test_export_chat_invite_link(bot: Bot, event_loop):
""" exportChatInviteLink method test """
from .types.dataset import CHAT, INVITE_LINK
chat = types.Chat(**CHAT)
async with FakeTelegram(message_dict=INVITE_LINK, loop=event_loop):
result = await bot.export_chat_invite_link(chat_id=chat.id)
assert result == INVITE_LINK
@pytest.mark.asyncio
async def test_delete_chat_photo(bot: Bot, event_loop):
""" deleteChatPhoto method test """
from .types.dataset import CHAT
chat = types.Chat(**CHAT)
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.delete_chat_photo(chat_id=chat.id)
assert isinstance(result, bool)
assert result is True
@pytest.mark.asyncio
async def test_set_chat_title(bot: Bot, event_loop):
""" setChatTitle method test """
from .types.dataset import CHAT
chat = types.Chat(**CHAT)
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.set_chat_title(chat_id=chat.id, title='Test title')
assert isinstance(result, bool)
assert result is True
@pytest.mark.asyncio
async def test_set_chat_description(bot: Bot, event_loop):
""" setChatDescription method test """
from .types.dataset import CHAT
chat = types.Chat(**CHAT)
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.set_chat_description(chat_id=chat.id, description='Test description')
assert isinstance(result, bool)
assert result is True
@pytest.mark.asyncio
async def test_pin_chat_message(bot: Bot, event_loop):
""" pinChatMessage method test """
from .types.dataset import MESSAGE
message = types.Message(**MESSAGE)
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.pin_chat_message(chat_id=message.chat.id, message_id=message.message_id,
disable_notification=False)
assert isinstance(result, bool)
assert result is True
@pytest.mark.asyncio
async def test_unpin_chat_message(bot: Bot, event_loop):
""" unpinChatMessage method test """
from .types.dataset import CHAT
chat = types.Chat(**CHAT)
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.unpin_chat_message(chat_id=chat.id)
assert isinstance(result, bool)
assert result is True
@pytest.mark.asyncio
async def test_leave_chat(bot: Bot, event_loop):
""" leaveChat method test """
from .types.dataset import CHAT
chat = types.Chat(**CHAT)
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.leave_chat(chat_id=chat.id)
assert isinstance(result, bool)
assert result is True
@pytest.mark.asyncio
async def test_get_chat(bot: Bot, event_loop):
""" getChat method test """
from .types.dataset import CHAT
chat = types.Chat(**CHAT)
async with FakeTelegram(message_dict=CHAT, loop=event_loop):
result = await bot.get_chat(chat_id=chat.id)
assert result == chat
@pytest.mark.asyncio
async def test_get_chat_administrators(bot: Bot, event_loop):
""" getChatAdministrators method test """
from .types.dataset import CHAT, CHAT_MEMBER
chat = types.Chat(**CHAT)
member = types.ChatMember(**CHAT_MEMBER)
async with FakeTelegram(message_dict=[CHAT_MEMBER, CHAT_MEMBER], loop=event_loop):
result = await bot.get_chat_administrators(chat_id=chat.id)
assert result[0] == member
assert len(result) == 2
@pytest.mark.asyncio
async def test_get_chat_members_count(bot: Bot, event_loop):
""" getChatMembersCount method test """
from .types.dataset import CHAT
chat = types.Chat(**CHAT)
count = 5
async with FakeTelegram(message_dict=count, loop=event_loop):
result = await bot.get_chat_members_count(chat_id=chat.id)
assert result == count
@pytest.mark.asyncio
async def test_get_chat_member(bot: Bot, event_loop):
""" getChatMember method test """
from .types.dataset import CHAT, CHAT_MEMBER
chat = types.Chat(**CHAT)
member = types.ChatMember(**CHAT_MEMBER)
async with FakeTelegram(message_dict=CHAT_MEMBER, loop=event_loop):
result = await bot.get_chat_member(chat_id=chat.id, user_id=member.user.id)
assert isinstance(result, types.ChatMember)
assert result == member
@pytest.mark.asyncio
async def test_set_chat_sticker_set(bot: Bot, event_loop):
""" setChatStickerSet method test """
from .types.dataset import CHAT
chat = types.Chat(**CHAT)
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.set_chat_sticker_set(chat_id=chat.id, sticker_set_name='aiogram_stickers')
assert isinstance(result, bool)
assert result is True
@pytest.mark.asyncio
async def test_delete_chat_sticker_set(bot: Bot, event_loop):
""" setChatStickerSet method test """
from .types.dataset import CHAT
chat = types.Chat(**CHAT)
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.delete_chat_sticker_set(chat_id=chat.id)
assert isinstance(result, bool)
assert result is True
@pytest.mark.asyncio
async def test_answer_callback_query(bot: Bot, event_loop):
""" answerCallbackQuery method test """
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.answer_callback_query(callback_query_id='QuERyId', text='Test Answer')
assert isinstance(result, bool)
assert result is True
@pytest.mark.asyncio
async def test_edit_message_text(bot: Bot, event_loop):
""" editMessageText method test """
from .types.dataset import EDITED_MESSAGE
msg = types.Message(**EDITED_MESSAGE)
# message by bot
async with FakeTelegram(message_dict=EDITED_MESSAGE, loop=event_loop):
result = await bot.edit_message_text(text=msg.text, chat_id=msg.chat.id, message_id=msg.message_id)
assert result == msg
# message by user
async with FakeTelegram(message_dict=True, loop=event_loop):
result = await bot.edit_message_text(text=msg.text, chat_id=msg.chat.id, message_id=msg.message_id)
assert isinstance(result, bool)
assert result is True

411
tests/types/dataset.py
View file

@ -1,10 +1,14 @@
""""
Dict data set for Telegram message types
"""
USER = {
"id": 12345678,
"is_bot": False,
"first_name": "FirstName",
"last_name": "LastName",
"username": "username",
"language_code": "ru-RU"
"language_code": "ru"
}
CHAT = {
@ -15,12 +19,38 @@ CHAT = {
"type": "private"
}
MESSAGE = {
"message_id": 11223,
"from": USER,
"chat": CHAT,
"date": 1508709711,
"text": "Hi, world!"
PHOTO = {
"file_id": "AgADBAADFak0G88YZAf8OAug7bHyS9x2ZxkABHVfpJywcloRAAGAAQABAg",
"file_size": 1101,
"width": 90,
"height": 51
}
AUDIO = {
"duration": 236,
"mime_type": "audio/mpeg3",
"title": "The Best Song",
"performer": "The Best Singer",
"file_id": "CQADAgADbQEAAsnrIUpNoRRNsH7_hAI",
"file_size": 9507774
}
CHAT_MEMBER = {
"user": USER,
"status": "administrator",
"can_be_edited": False,
"can_change_info": True,
"can_delete_messages": True,
"can_invite_users": True,
"can_restrict_members": True,
"can_pin_messages": True,
"can_promote_members": False
}
CONTACT = {
"phone_number": "88005553535",
"first_name": "John",
"last_name": "Smith",
}
DOCUMENT = {
@ -30,27 +60,6 @@ DOCUMENT = {
"file_size": 21331
}
MESSAGE_WITH_DOCUMENT = {
"message_id": 12345,
"from": USER,
"chat": CHAT,
"date": 1508768012,
"document": DOCUMENT,
"caption": "doc description"
}
UPDATE = {
"update_id": 128526,
"message": MESSAGE
}
PHOTO = {
"file_id": "AgADBAADFak0G88YZAf8OAug7bHyS9x2ZxkABHVfpJywcloRAAGAAQABAg",
"file_size": 1101,
"width": 90,
"height": 51
}
ANIMATION = {
"file_name": "a9b0e0ca537aa344338f80978f0896b7.gif.mp4",
"mime_type": "video/mp4",
@ -59,6 +68,43 @@ ANIMATION = {
"file_size": 65837
}
ENTITY_BOLD = {
"offset": 5,
"length": 2,
"type": "bold"
}
ENTITY_ITALIC = {
"offset": 8,
"length": 1,
"type": "italic"
}
ENTITY_LINK = {
"offset": 10,
"length": 6,
"type": "text_link",
"url": "http://google.com/"
}
ENTITY_CODE = {
"offset": 17,
"length": 7,
"type": "code"
}
ENTITY_PRE = {
"offset": 30,
"length": 4,
"type": "pre"
}
ENTITY_MENTION = {
"offset": 47,
"length": 9,
"type": "mention"
}
GAME = {
"title": "Karate Kido",
"description": "No trees were harmed in the making of this game :)",
@ -66,6 +112,162 @@ GAME = {
"animation": ANIMATION
}
INVOICE = {
"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!",
"start_parameter": "time-machine-example",
"currency": "USD",
"total_amount": 6250
}
LOCATION = {
"latitude": 50.693416,
"longitude": 30.624605
}
VENUE = {
"location": LOCATION,
"title": "Venue Name",
"address": "Venue Address",
"foursquare_id": "4e6f2cec483bad563d150f98"
}
SHIPPING_ADDRESS = {
"country_code": "US",
"state": "State",
"city": "DefaultCity",
"street_line1": "Central",
"street_line2": "Middle",
"post_code": "424242"
}
STICKER = {
"width": 512,
"height": 512,
"emoji": "🛠",
"set_name": "StickerSet",
"thumb": {
"file_id": "AAbbCCddEEffGGhh1234567890",
"file_size": 1234,
"width": 128,
"height": 128
},
"file_id": "AAbbCCddEEffGGhh1234567890",
"file_size": 12345
}
SUCCESSFUL_PAYMENT = {
"currency": "USD",
"total_amount": 6250,
"invoice_payload": "HAPPY FRIDAYS COUPON",
"telegram_payment_charge_id": "_",
"provider_payment_charge_id": "12345678901234_test"
}
VIDEO = {
"duration": 52,
"width": 853,
"height": 480,
"mime_type": "video/quicktime",
"thumb": PHOTO,
"file_id": "BAADAgpAADdawy_JxS72kRvV3cortAg",
"file_size": 10099782
}
VIDEO_NOTE = {
"duration": 4,
"length": 240,
"thumb": PHOTO,
"file_id": "AbCdEfGhIjKlMnOpQrStUvWxYz",
"file_size": 186562
}
VOICE = {
"duration": 1,
"mime_type": "audio/ogg",
"file_id": "AwADawAgADADy_JxS2gopIVIIxlhAg",
"file_size": 4321
}
CALLBACK_QUERY = {}
CHANNEL_POST = {}
CHOSEN_INLINE_RESULT = {}
EDITED_CHANNEL_POST = {}
EDITED_MESSAGE = {
"message_id": 12345,
"from": USER,
"chat": CHAT,
"date": 1508825372,
"edit_date": 1508825379,
"text": "hi there (edited)"
}
FORWARDED_MESSAGE = {
"message_id": 12345,
"from": USER,
"chat": CHAT,
"date": 1522828529,
"forward_from_chat": CHAT,
"forward_from_message_id": 123,
"forward_date": 1522749037,
"text": "Forwarded text with entities from public channel ",
"entities": [ENTITY_BOLD, ENTITY_CODE, ENTITY_ITALIC, ENTITY_LINK,
ENTITY_LINK, ENTITY_MENTION, ENTITY_PRE]
}
INLINE_QUERY = {}
MESSAGE = {
"message_id": 11223,
"from": USER,
"chat": CHAT,
"date": 1508709711,
"text": "Hi, world!"
}
MESSAGE_WITH_AUDIO = {
"message_id": 12345,
"from": USER,
"chat": CHAT,
"date": 1508739776,
"audio": AUDIO,
"caption": "This is my favourite song"
}
MESSAGE_WITH_AUTHOR_SIGNATURE = {}
MESSAGE_WITH_CHANNEL_CHAT_CREATED = {}
MESSAGE_WITH_CONTACT = {
"message_id": 56006,
"from": USER,
"chat": CHAT,
"date": 1522850298,
"contact": CONTACT
}
MESSAGE_WITH_DELETE_CHAT_PHOTO = {}
MESSAGE_WITH_DOCUMENT = {
"message_id": 12345,
"from": USER,
"chat": CHAT,
"date": 1508768012,
"document": DOCUMENT,
"caption": "Read my document"
}
MESSAGE_WITH_EDIT_DATE = {}
MESSAGE_WITH_ENTITIES = {}
MESSAGE_WITH_GAME = {
"message_id": 12345,
"from": USER,
@ -73,3 +275,156 @@ MESSAGE_WITH_GAME = {
"date": 1508824810,
"game": GAME
}
MESSAGE_WITH_GROUP_CHAT_CREATED = {}
MESSAGE_WITH_INVOICE = {
"message_id": 9772,
"from": USER,
"chat": CHAT,
"date": 1508761719,
"invoice": INVOICE
}
MESSAGE_WITH_LEFT_CHAT_MEMBER = {}
MESSAGE_WITH_LOCATION = {
"message_id": 12345,
"from": USER,
"chat": CHAT,
"date": 1508755473,
"location": LOCATION
}
MESSAGE_WITH_MIGRATE_FROM_CHAT_ID = {}
MESSAGE_WITH_MIGRATE_TO_CHAT_ID = {}
MESSAGE_WITH_NEW_CHAT_MEMBERS = {}
MESSAGE_WITH_NEW_CHAT_PHOTO = {}
MESSAGE_WITH_NEW_CHAT_TITLE = {}
MESSAGE_WITH_PHOTO = {
"message_id": 12345,
"from": USER,
"chat": CHAT,
"date": 1508825154,
"photo": [PHOTO, PHOTO, PHOTO, PHOTO],
"caption": "photo description"
}
MESSAGE_WITH_MEDIA_GROUP = {
"message_id": 55966,
"from": USER,
"chat": CHAT,
"date": 1522843665,
"media_group_id": "12182749320567362",
"photo": [PHOTO, PHOTO, PHOTO, PHOTO]
}
MESSAGE_WITH_PINNED_MESSAGE = {}
MESSAGE_WITH_REPLY_TO_MESSAGE = {}
MESSAGE_WITH_STICKER = {
"message_id": 12345,
"from": USER,
"chat": CHAT,
"date": 1508771450,
"sticker": STICKER
}
MESSAGE_WITH_SUCCESSFUL_PAYMENT = {
"message_id": 9768,
"from": USER,
"chat": CHAT,
"date": 1508761169,
"successful_payment": SUCCESSFUL_PAYMENT
}
MESSAGE_WITH_SUPERGROUP_CHAT_CREATED = {}
MESSAGE_WITH_VENUE = {
"message_id": 56004,
"from": USER,
"chat": CHAT,
"date": 1522849819,
"location": LOCATION,
"venue": VENUE
}
MESSAGE_WITH_VIDEO = {
"message_id": 12345,
"from": USER,
"chat": CHAT,
"date": 1508756494,
"video": VIDEO,
"caption": "description"
}
MESSAGE_WITH_VIDEO_NOTE = {
"message_id": 55934,
"from": USER,
"chat": CHAT,
"date": 1522835890,
"video_note": VIDEO_NOTE
}
MESSAGE_WITH_VOICE = {
"message_id": 12345,
"from": USER,
"chat": CHAT,
"date": 1508768403,
"voice": VOICE
}
PRE_CHECKOUT_QUERY = {
"id": "262181558630368727",
"from": USER,
"currency": "USD",
"total_amount": 6250,
"invoice_payload": "HAPPY FRIDAYS COUPON"
}
REPLY_MESSAGE = {
"message_id": 12345,
"from": USER,
"chat": CHAT,
"date": 1508751866,
"reply_to_message": MESSAGE,
"text": "Reply to quoted message"
}
SHIPPING_QUERY = {
"id": "262181558684397422",
"from": USER,
"invoice_payload": "HAPPY FRIDAYS COUPON",
"shipping_address": SHIPPING_ADDRESS
}
USER_PROFILE_PHOTOS = {
"total_count": 1, "photos": [
[PHOTO, PHOTO, PHOTO]
]
}
FILE = {
"file_id": "XXXYYYZZZ",
"file_size": 5254,
"file_path": "voice\/file_8"
}
INVITE_LINK = 'https://t.me/joinchat/AbCdEfjKILDADwdd123'
UPDATE = {
"update_id": 123456789,
"message": MESSAGE
}
WEBHOOK_INFO = {
"url": "",
"has_custom_certificate": False,
"pending_update_count": 0
}