python/aiogram
1
0
Fork 0
mirror of https://github.com/aiogram/aiogram.git synced 2025-12-11 01:54:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'dev-1.x-upstream' into dev-1.x

Browse source
This commit is contained in:
Suren Khorenyan 2018-05-20 20:52:44 +03:00
commit 5116ce83fd
35 changed files with 1885 additions and 482 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
LICENSE
View file

@ -1,4 +1,4 @@
Copyright (c) 2017 Alex Root Junior
Copyright (c) 2017-2018 Alex Root Junior
Permission is hereby granted, free of charge, to any person obtaining a copy of this
software and associated documentation files (the "Software"), to deal in the Software

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)'"

25
aiogram/__init__.py
View file

@ -1,27 +1,14 @@
import warnings
import asyncio
try:
from .bot import Bot
except ImportError as e:
if e.name == 'aiohttp':
warnings.warn('Dependencies are not installed!',
category=ImportWarning)
else:
raise
from .utils.versions import Stage, Version
from .bot import Bot
from .dispatcher import Dispatcher
try:
import uvloop
except ImportError:
pass
uvloop = None
else:
import asyncio
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
VERSION = Version(1, 2, 1, stage=Stage.DEV, build=0)
API_VERSION = Version(3, 6)
__version__ = VERSION.version
__api_version__ = API_VERSION.version
__version__ = '1.3.1.dev1'
__api_version__ = '3.6'

25
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
@ -59,23 +59,28 @@ async def _check_result(method_name, response):
result_json = {}
description = result_json.get('description') or body
parameters = types.ResponseParameters(**result_json.get('parameters', {}) or {})
if HTTPStatus.OK <= response.status <= HTTPStatus.IM_USED:
return result_json.get('result')
elif 'retry_after' in result_json:
raise exceptions.RetryAfter(result_json['retry_after'])
elif 'migrate_to_chat_id' in result_json:
raise exceptions.MigrateToChat(result_json['migrate_to_chat_id'])
elif parameters.retry_after:
raise exceptions.RetryAfter(parameters.retry_after)
elif parameters.migrate_to_chat_id:
raise exceptions.MigrateToChat(parameters.migrate_to_chat_id)
elif response.status == HTTPStatus.BAD_REQUEST:
raise exceptions.BadRequest(description)
exceptions.BadRequest.detect(description)
elif response.status == HTTPStatus.NOT_FOUND:
exceptions.NotFound.detect(description)
elif response.status == HTTPStatus.CONFLICT:
raise exceptions.ConflictError(description)
exceptions.ConflictError.detect(description)
elif response.status in [HTTPStatus.UNAUTHORIZED, HTTPStatus.FORBIDDEN]:
raise exceptions.Unauthorized(description)
exceptions.Unauthorized.detect(description)
elif response.status == HTTPStatus.REQUEST_ENTITY_TOO_LARGE:
raise exceptions.NetworkError('File too large for uploading. '
'Check telegram api limits https://core.telegram.org/bots/api#senddocument')
elif response.status >= HTTPStatus.INTERNAL_SERVER_ERROR:
if 'restart' in description:
raise exceptions.RestartingTelegram()
raise exceptions.TelegramAPIError(description)
raise exceptions.TelegramAPIError(f"{description} [{response.status}]")
@ -161,7 +166,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

80
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
@ -17,7 +19,7 @@ class BaseBot:
def __init__(self, token: base.String,
loop: Optional[Union[asyncio.BaseEventLoop, asyncio.AbstractEventLoop]] = None,
connections_limit: Optional[base.Integer] = 10,
connections_limit: Optional[base.Integer] = None,
proxy: Optional[base.String] = None, proxy_auth: Optional[aiohttp.BasicAuth] = None,
validate_token: Optional[base.Boolean] = True,
parse_mode=None):
@ -55,12 +57,19 @@ 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())
# Temp sessions
self._temp_sessions = []
if isinstance(proxy, str) and proxy.startswith('socks5://'):
from aiosocksy.connector import ProxyClientRequest, ProxyConnector
connector = ProxyConnector(limit=connections_limit, ssl_context=ssl_context, loop=self.loop)
request_class = ProxyClientRequest
else:
connector = aiohttp.TCPConnector(limit=connections_limit, ssl_context=ssl_context,
loop=self.loop)
request_class = aiohttp.ClientRequest
self.session = aiohttp.ClientSession(connector=connector, request_class=request_class,
loop=self.loop, json_serialize=json.dumps)
# Data stored in bot instance
self._data = {}
@ -68,7 +77,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):
"""
@ -76,38 +86,6 @@ class BaseBot:
"""
if self.session and not self.session.closed:
await self.session.close()
for session in self._temp_sessions:
if not session.closed:
await session.close()
def create_temp_session(self, limit: base.Integer = 1, force_close: base.Boolean = False) -> aiohttp.ClientSession:
"""
Create temporary session
:param limit: Limit of connections
:type limit: :obj:`int`
:param force_close: Set to True to force close and do reconnect after each request (and between redirects).
:type force_close: :obj:`bool`
:return: New session
:rtype: :obj:`aiohttp.TCPConnector`
"""
session = aiohttp.ClientSession(
connector=aiohttp.TCPConnector(limit=limit, force_close=force_close),
loop=self.loop, json_serialize=json.dumps)
self._temp_sessions.append(session)
return session
def destroy_temp_session(self, session: aiohttp.ClientSession):
"""
Destroy temporary session
:param session: target session
:type session: :obj:`aiohttp.ClientSession`
"""
if not session.closed:
session.close()
if session in self._temp_sessions:
self._temp_sessions.remove(session)
async def request(self, method: base.String,
data: Optional[Dict] = None,
@ -152,23 +130,19 @@ class BaseBot:
if destination is None:
destination = io.BytesIO()
session = self.create_temp_session()
url = api.Methods.file_url(token=self.__token, path=file_path)
dest = destination if isinstance(destination, io.IOBase) else open(destination, 'wb')
try:
async with session.get(url, timeout=timeout, proxy=self.proxy, proxy_auth=self.proxy_auth) as response:
while True:
chunk = await response.content.read(chunk_size)
if not chunk:
break
dest.write(chunk)
dest.flush()
if seek:
dest.seek(0)
return dest
finally:
self.destroy_temp_session(session)
async with self.session.get(url, timeout=timeout, proxy=self.proxy, proxy_auth=self.proxy_auth) as response:
while True:
chunk = await response.content.read(chunk_size)
if not chunk:
break
dest.write(chunk)
dest.flush()
if seek:
dest.seek(0)
return dest
async def send_file(self, file_type, method, file, payload) -> Union[Dict, base.Boolean]:
"""

18
aiogram/bot/bot.py
View file

@ -249,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
@ -295,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
@ -343,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
@ -394,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.
@ -441,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.
@ -1305,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,

13
aiogram/contrib/fsm_storage/rethinkdb.py
View file

@ -11,6 +11,9 @@ __all__ = ['RethinkDBStorage', 'ConnectionNotClosed']
r.set_loop_type('asyncio')
# TODO: rewrite connections pool
class ConnectionNotClosed(Exception):
"""
Indicates that DB connection wasn't closed.
@ -101,19 +104,17 @@ class RethinkDBStorage(BaseStorage):
except r.ReqlError:
raise ConnectionNotClosed('Exception was caught while closing connection')
def wait_closed(self):
async def wait_closed(self):
"""
Checks if connection is closed.
Does nothing
"""
if len(self._outstanding_connections) != 0 and self._queue.qsize() != 0:
raise ConnectionNotClosed
return True
pass
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]:
chat, user = map(str, self.check_address(chat=chat, user=user))
conn = await self.get_connection()
result = await r.table(self._table).get(chat)[user]['state'].default(default or '').run(conn)
result = await r.table(self._table).get(chat)[user]['state'].default(default or None).run(conn)
await self.put_connection(conn)
return result

20
aiogram/dispatcher/__init__.py
View file

@ -15,7 +15,6 @@ from .webhook import BaseResponse
from ..bot import Bot
from ..types.message import ContentType
from ..utils import context
from ..utils.deprecated import deprecated
from ..utils.exceptions import NetworkError, TelegramAPIError, Throttled
log = logging.getLogger(__name__)
@ -200,17 +199,6 @@ class Dispatcher:
return await self.bot.delete_webhook()
@deprecated('The old method was renamed to `start_polling`')
async def start_pooling(self, *args, **kwargs):
"""
Start long-lopping
:param args:
:param kwargs:
:return:
"""
return await self.start_polling(*args, **kwargs)
async def start_polling(self, timeout=20, relax=0.1, limit=None, reset_webhook=None):
"""
Start long-polling
@ -276,10 +264,6 @@ class Dispatcher:
except TelegramAPIError:
log.exception('Cause exception while processing updates.')
@deprecated('The old method was renamed to `stop_polling`')
def stop_pooling(self):
return self.stop_polling()
def stop_polling(self):
"""
Break long-polling process.
@ -298,10 +282,6 @@ class Dispatcher:
"""
await asyncio.shield(self._close_waiter, loop=self.loop)
@deprecated('The old method was renamed to `is_polling`')
def is_pooling(self):
return self.is_polling()
def is_polling(self):
"""
Check if polling is enabled

13
aiogram/dispatcher/filters.py
View file

@ -2,7 +2,7 @@ import asyncio
import inspect
import re
from ..types import ContentType
from ..types import CallbackQuery, ContentType, Message
from ..utils import context
from ..utils.helper import Helper, HelperMode, Item
@ -127,9 +127,12 @@ class RegexpFilter(Filter):
def __init__(self, regexp):
self.regexp = re.compile(regexp, flags=re.IGNORECASE | re.MULTILINE)
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):
@ -168,7 +171,7 @@ class ContentTypeFilter(Filter):
def check(self, message):
return ContentType.ANY[0] in self.content_types or \
message.content_type in self.content_types
message.content_type in self.content_types
class CancelFilter(Filter):

2
aiogram/dispatcher/handler.py
View file

@ -1,5 +1,5 @@
from aiogram.utils import context
from .filters import check_filters
from ..utils import context
class SkipHandler(BaseException):

16
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'
@ -324,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)

7
aiogram/dispatcher/webhook.py
View file

@ -87,8 +87,11 @@ class WebhookRequestHandler(web.View):
:return: :class:`aiogram.Dispatcher`
"""
dp = self.request.app[BOT_DISPATCHER_KEY]
context.set_value('dispatcher', dp)
context.set_value('bot', dp.bot)
try:
context.set_value('dispatcher', dp)
context.set_value('bot', dp.bot)
except RuntimeError:
pass
return dp
async def parse_update(self, bot):

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)

2
aiogram/types/inline_keyboard.py
View file

@ -69,7 +69,7 @@ class InlineKeyboardMarkup(base.TelegramObject):
: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)

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!')

9
aiogram/types/input_message_content.py
View file

@ -62,9 +62,18 @@ class InputTextMessageContent(InputMessageContent):
parse_mode: base.String = fields.Field()
disable_web_page_preview: base.Boolean = fields.Field()
def safe_get_parse_mode(self):
try:
return self.bot.parse_mode
except RuntimeError:
pass
def __init__(self, message_text: typing.Optional[base.String] = None,
parse_mode: typing.Optional[base.String] = None,
disable_web_page_preview: typing.Optional[base.Boolean] = None):
if parse_mode is None:
parse_mode = self.safe_get_parse_mode()
super(InputTextMessageContent, self).__init__(message_text=message_text, parse_mode=parse_mode,
disable_web_page_preview=disable_web_page_preview)

4
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:

34
aiogram/utils/auth_widget.py
View file

@ -1,14 +1,18 @@
import collections
"""
Implementation of Telegram site authorization checking mechanism
for more information https://core.telegram.org/widgets/login#checking-authorization
Source: https://gist.github.com/JrooTJunior/887791de7273c9df5277d2b1ecadc839
"""
import hashlib
import hmac
import collections
def check_token(data, token):
def generate_hash(data: dict, token: str) -> str:
"""
Validate auth token
https://core.telegram.org/widgets/login#checking-authorization
Source: https://gist.github.com/xen/e4bea72487d34caa28c762776cf655a3
Generate secret hash
:param data:
:param token:
@ -17,9 +21,17 @@ def check_token(data, token):
secret = hashlib.sha256()
secret.update(token.encode('utf-8'))
sorted_params = collections.OrderedDict(sorted(data.items()))
param_hash = sorted_params.pop('hash', '') or ''
msg = "\n".join(["{}={}".format(k, v) for k, v in sorted_params.items()])
msg = '\n'.join(["{}={}".format(k, v) for k, v in sorted_params.items() if k != 'hash'])
return hmac.new(secret.digest(), msg.encode('utf-8'), digestmod=hashlib.sha256).hexdigest()
if param_hash == hmac.new(secret.digest(), msg.encode('utf-8'), digestmod=hashlib.sha256).hexdigest():
return True
return False
def check_token(data: dict, token: str) -> bool:
"""
Validate auth token
:param data:
:param token:
:return:
"""
param_hash = data.get('hash', '') or ''
return param_hash == generate_hash(data, token)

2
aiogram/utils/context.py
View file

@ -46,6 +46,8 @@ def get_current_state() -> typing.Dict:
:rtype: :obj:`dict`
"""
task = asyncio.Task.current_task()
if task is None:
raise RuntimeError('Can be used only in Task context.')
context_ = getattr(task, 'context', None)
if context_ is None:
context_ = task.context = {}

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)

269
aiogram/utils/exceptions.py
View file

@ -1,6 +1,53 @@
"""
TelegramAPIError
ValidationError
Throttled
BadRequest
MessageError
MessageNotModified
MessageToForwardNotFound
MessageToDeleteNotFound
MessageIdentifierNotSpecified
MessageTextIsEmpty
ToMuchMessages
ChatNotFound
InvalidQueryID
InvalidPeerID
InvalidHTTPUrlContent
WrongFileIdentifier
GroupDeactivated
BadWebhook
WebhookRequireHTTPS
BadWebhookPort
BadWebhookAddrInfo
CantParseUrl
NotFound
MethodNotKnown
PhotoAsInputFileRequired
InvalidStickersSet
ChatAdminRequired
PhotoDimensions
UnavailableMembers
TypeOfFileMismatch
ConflictError
TerminatedByOtherGetUpdates
CantGetUpdates
Unauthorized
BotKicked
BotBlocked
UserDeactivated
CantInitiateConversation
NetworkError
RetryAfter
MigrateToChat
RestartingTelegram
AIOGramWarning
TimeoutWarning
"""
import time
_PREFIXES = ['Error: ', '[Error]: ', 'Bad Request: ', 'Conflict: ']
_PREFIXES = ['Error: ', '[Error]: ', 'Bad Request: ', 'Conflict: ', 'Not Found: ']
def _clean_message(text):
@ -11,10 +58,52 @@ 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
__subclasses = []
def __init_subclass__(cls, **kwargs):
super(_MatchErrorMixin, cls).__init_subclass__(**kwargs)
# cls.match = cls.match.lower() if cls.match else ''
if not hasattr(cls, f"_{cls.__name__}__group"):
cls.__subclasses.append(cls)
@classmethod
def check(cls, message) -> bool:
"""
Compare pattern with message
:param message: always must be in lowercase
:return: bool
"""
return cls.match.lower() in message
@classmethod
def throw(cls):
"""
Throw error
:raise: this
"""
raise cls(cls.text or cls.match)
@classmethod
def detect(cls, description):
description = description.lower()
for err in cls.__subclasses:
if err is cls:
continue
if err.check(description):
err.throw()
raise cls(description)
class AIOGramWarning(Warning):
pass
@ -23,26 +112,188 @@ class TimeoutWarning(AIOGramWarning):
pass
class FSMStorageWarning(AIOGramWarning):
pass
class ValidationError(TelegramAPIError):
pass
class BadRequest(TelegramAPIError):
pass
class BadRequest(TelegramAPIError, _MatchErrorMixin):
__group = True
class ConflictError(TelegramAPIError):
pass
class MessageError(BadRequest):
__group = True
class Unauthorized(TelegramAPIError):
pass
class MessageNotModified(MessageError):
"""
Will be raised when you try to set new text is equals to current text.
"""
match = 'message is not modified'
class MessageToForwardNotFound(MessageError):
"""
Will be raised when you try to forward very old or deleted or unknown message.
"""
match = 'message to forward not found'
class MessageToDeleteNotFound(MessageError):
"""
Will be raised when you try to delete very old or deleted or unknown message.
"""
match = 'message to delete not found'
class MessageIdentifierNotSpecified(MessageError):
match = 'message identifier is not specified'
class MessageTextIsEmpty(MessageError):
match = 'Message text is empty'
class ToMuchMessages(MessageError):
"""
Will be raised when you try to send media group with more than 10 items.
"""
match = 'Too much messages to send as an album'
class ChatNotFound(BadRequest):
match = 'chat not found'
class InvalidQueryID(BadRequest):
match = 'QUERY_ID_INVALID'
text = 'Invalid query ID'
class InvalidPeerID(BadRequest):
match = 'PEER_ID_INVALID'
text = 'Invalid peer ID'
class InvalidHTTPUrlContent(BadRequest):
match = 'Failed to get HTTP URL content'
class WrongFileIdentifier(BadRequest):
match = 'wrong file identifier/HTTP URL specified'
class GroupDeactivated(BadRequest):
match = 'group is deactivated'
class PhotoAsInputFileRequired(BadRequest):
"""
Will be raised when you try to set chat photo from file ID.
"""
match = 'Photo should be uploaded as an InputFile'
class InvalidStickersSet(BadRequest):
match = 'STICKERSET_INVALID'
text = 'Stickers set is invalid'
class ChatAdminRequired(BadRequest):
match = 'CHAT_ADMIN_REQUIRED'
text = 'Admin permissions is required!'
class PhotoDimensions(BadRequest):
match = 'PHOTO_INVALID_DIMENSIONS'
text = 'Invalid photo dimensions'
class UnavailableMembers(BadRequest):
match = 'supergroup members are unavailable'
class TypeOfFileMismatch(BadRequest):
match = 'type of file mismatch'
class BadWebhook(BadRequest):
__group = True
class WebhookRequireHTTPS(BadWebhook):
match = 'HTTPS url must be provided for webhook'
text = 'bad webhook: ' + match
class BadWebhookPort(BadWebhook):
match = 'Webhook can be set up only on ports 80, 88, 443 or 8443'
text = 'bad webhook: ' + match
class BadWebhookAddrInfo(BadWebhook):
match = 'getaddrinfo: Temporary failure in name resolution'
text = 'bad webhook: ' + match
class CantParseUrl(BadRequest):
match = 'can\'t parse URL'
class NotFound(TelegramAPIError, _MatchErrorMixin):
__group = True
class MethodNotKnown(NotFound):
match = 'method not found'
class ConflictError(TelegramAPIError, _MatchErrorMixin):
__group = True
class TerminatedByOtherGetUpdates(ConflictError):
match = 'terminated by other getUpdates request'
text = 'Terminated by other getUpdates request; ' \
'Make sure that only one bot instance is running'
class CantGetUpdates(ConflictError):
match = 'can\'t use getUpdates method while webhook is active'
class Unauthorized(TelegramAPIError, _MatchErrorMixin):
__group = True
class BotKicked(Unauthorized):
match = 'Bot was kicked from a chat'
class BotBlocked(Unauthorized):
match = 'bot was blocked by the user'
class UserDeactivated(Unauthorized):
match = 'user is deactivated'
class CantInitiateConversation(Unauthorized):
match = 'bot can\'t initiate conversation with a user'
class NetworkError(TelegramAPIError):
pass
class RestartingTelegram(TelegramAPIError):
def __init__(self):
super(RestartingTelegram, self).__init__('The Telegram Bot API service is restarting. Wait few second.')
class RetryAfter(TelegramAPIError):
def __init__(self, retry_after):
super(RetryAfter, self).__init__(f"Flood control exceeded. Retry in {retry_after} seconds.")
@ -55,7 +306,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>')

371
aiogram/utils/executor.py
View file

@ -1,93 +1,316 @@
import asyncio
import datetime
import functools
import secrets
from warnings import warn
from aiohttp import web
from . import context
from .deprecated import deprecated
from ..bot.api import log
from ..dispatcher import Dispatcher
from ..dispatcher.webhook import BOT_DISPATCHER_KEY, get_new_configured_app
from ..dispatcher.webhook import BOT_DISPATCHER_KEY, WebhookRequestHandler
APP_EXECUTOR_KEY = 'APP_EXECUTOR'
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.")
def _setup_callbacks(executor, on_startup=None, on_shutdown=None):
if on_startup is not None:
executor.on_startup(on_startup)
if on_shutdown is not None:
executor.on_shutdown(on_shutdown)
async def _wh_startup(app):
callback = app.get('_startup_callback', None)
dispatcher = app.get(BOT_DISPATCHER_KEY, None)
skip_updates = app.get('_skip_updates', False)
await _startup(dispatcher, skip_updates=skip_updates, callback=callback)
async def _shutdown(dispatcher: Dispatcher, callback=None):
if callable(callback):
await callback(dispatcher)
if dispatcher.is_polling():
dispatcher.stop_polling()
# await dispatcher.wait_closed()
await dispatcher.storage.close()
await dispatcher.storage.wait_closed()
await dispatcher.bot.close()
async def _wh_shutdown(app):
callback = app.get('_shutdown_callback', None)
dispatcher = app.get(BOT_DISPATCHER_KEY, None)
await _shutdown(dispatcher, callback=callback)
@deprecated('The old function was renamed to `start_polling`')
def start_pooling(*args, **kwargs):
return start_polling(*args, **kwargs)
def start_polling(dispatcher, *, loop=None, skip_updates=False,
def start_polling(dispatcher, *, loop=None, skip_updates=False, reset_webhook=True,
on_startup=None, on_shutdown=None):
log.warning('Start bot with long-polling.')
if loop is None:
loop = dispatcher.loop
"""
Start bot in long-polling mode
loop.set_task_factory(context.task_factory)
:param dispatcher:
:param loop:
:param skip_updates:
:param reset_webhook:
:param on_startup:
:param on_shutdown:
"""
executor = Executor(dispatcher, skip_updates=skip_updates, loop=loop)
_setup_callbacks(executor, on_startup, on_shutdown)
try:
loop.run_until_complete(_startup(dispatcher, skip_updates, on_startup))
loop.create_task(dispatcher.start_polling(reset_webhook=True))
loop.run_forever()
except (KeyboardInterrupt, SystemExit):
pass
finally:
loop.run_until_complete(_shutdown(dispatcher, callback=on_shutdown))
log.warning("Goodbye!")
executor.start_polling(reset_webhook=reset_webhook)
def start_webhook(dispatcher, webhook_path, *, loop=None, skip_updates=None,
on_startup=None, on_shutdown=None, check_ip=False, **kwargs):
log.warning('Start bot with webhook.')
if loop is None:
loop = dispatcher.loop
"""
Start bot in webhook mode
loop.set_task_factory(context.task_factory)
:param dispatcher:
:param webhook_path:
:param loop:
:param skip_updates:
:param on_startup:
:param on_shutdown:
:param check_ip:
:param kwargs:
:return:
"""
executor = Executor(dispatcher, skip_updates=skip_updates, check_ip=check_ip, loop=loop)
_setup_callbacks(executor, on_startup, on_shutdown)
app = get_new_configured_app(dispatcher, webhook_path)
app['_startup_callback'] = on_startup
app['_shutdown_callback'] = on_shutdown
app['_skip_updates'] = skip_updates
app['_check_ip'] = check_ip
executor.start_webhook(webhook_path, **kwargs)
app.on_startup.append(_wh_startup)
app.on_shutdown.append(_wh_shutdown)
web.run_app(app, loop=loop, **kwargs)
return app
def start(dispatcher, future, *, loop=None, skip_updates=None,
on_startup=None, on_shutdown=None):
"""
Execute Future.
:param dispatcher: instance of Dispatcher
:param future: future
:param loop: instance of AbstractEventLoop
:param skip_updates:
:param on_startup:
:param on_shutdown:
:return:
"""
executor = Executor(dispatcher, skip_updates=skip_updates, loop=loop)
_setup_callbacks(executor, on_startup, on_shutdown)
return executor.start(future)
class Executor:
"""
Main executor class
"""
def __init__(self, dispatcher, skip_updates=None, check_ip=False, loop=None):
if loop is None:
loop = dispatcher.loop
self.dispatcher = dispatcher
self.skip_updates = skip_updates
self.check_ip = check_ip
self.loop = loop
self._identity = secrets.token_urlsafe(16)
self._web_app = None
self._on_startup_webhook = []
self._on_startup_polling = []
self._on_shutdown_webhook = []
self._on_shutdown_polling = []
self._freeze = False
@property
def frozen(self):
return self._freeze
def set_web_app(self, application: web.Application):
"""
Change instance of aiohttp.web.Applicaton
:param application:
"""
self._web_app = application
@property
def web_app(self) -> web.Application:
if self._web_app is None:
raise RuntimeError('web.Application() is not configured!')
return self._web_app
def on_startup(self, callback: callable, polling=True, webhook=True):
"""
Register a callback for the startup process
:param callback:
:param polling: use with polling
:param webhook: use with webhook
"""
self._check_frozen()
if not webhook and not polling:
warn('This action has no effect!', UserWarning)
return
if isinstance(callback, (list, tuple, set)):
for cb in callback:
self.on_startup(cb, polling, webhook)
return
if polling:
self._on_startup_polling.append(callback)
if webhook:
self._on_startup_webhook.append(callback)
def on_shutdown(self, callback: callable, polling=True, webhook=True):
"""
Register a callback for the shutdown process
:param callback:
:param polling: use with polling
:param webhook: use with webhook
"""
self._check_frozen()
if not webhook and not polling:
warn('This action has no effect!', UserWarning)
return
if isinstance(callback, (list, tuple, set)):
for cb in callback:
self.on_shutdown(cb, polling, webhook)
return
if polling:
self._on_shutdown_polling.append(callback)
if webhook:
self._on_shutdown_webhook.append(callback)
def _check_frozen(self):
if self.frozen:
raise RuntimeError('Executor is frozen!')
def _prepare_polling(self):
self._check_frozen()
self._freeze = True
self.loop.set_task_factory(context.task_factory)
def _prepare_webhook(self, path=None, handler=WebhookRequestHandler):
self._check_frozen()
self._freeze = True
self.loop.set_task_factory(context.task_factory)
app = self._web_app
if app is None:
self._web_app = app = web.Application()
if self._identity == app.get(self._identity):
# App is already configured
return
if path is not None:
app.router.add_route('*', path, handler, name='webhook_handler')
async def _wrap_callback(cb, _):
return await cb(self.dispatcher)
for callback in self._on_startup_webhook:
app.on_startup.append(functools.partial(_wrap_callback, callback))
# for callback in self._on_shutdown_webhook:
# app.on_shutdown.append(functools.partial(_wrap_callback, callback))
async def _on_shutdown(_):
await self._shutdown_webhook()
app.on_shutdown.append(_on_shutdown)
app[APP_EXECUTOR_KEY] = self
app[BOT_DISPATCHER_KEY] = self.dispatcher
app[self._identity] = datetime.datetime.now()
app['_check_ip'] = self.check_ip
def start_webhook(self, webhook_path=None, request_handler=WebhookRequestHandler, **kwargs):
"""
Start bot in webhook mode
:param webhook_path:
:param request_handler:
:param kwargs:
:return:
"""
self._prepare_webhook(webhook_path, request_handler)
self.loop.run_until_complete(self._startup_webhook())
web.run_app(self._web_app, **kwargs)
def start_polling(self, reset_webhook=None):
"""
Start bot in long-polling mode
:param reset_webhook:
"""
self._prepare_polling()
loop: asyncio.AbstractEventLoop = self.loop
try:
loop.run_until_complete(self._startup_polling())
loop.create_task(self.dispatcher.start_polling(reset_webhook=reset_webhook))
loop.run_forever()
except (KeyboardInterrupt, SystemExit):
# loop.stop()
pass
finally:
loop.run_until_complete(self._shutdown_polling())
log.warning("Goodbye!")
def start(self, future):
"""
Execute Future.
Return the Future's result, or raise its exception.
:param future:
:return:
"""
self._check_frozen()
self._freeze = True
loop: asyncio.AbstractEventLoop = self.loop
try:
loop.run_until_complete(self._startup_polling())
result = loop.run_until_complete(future)
except (KeyboardInterrupt, SystemExit):
result = None
loop.stop()
finally:
loop.run_until_complete(self._shutdown_polling())
log.warning("Goodbye!")
return result
async def _skip_updates(self):
await self.dispatcher.reset_webhook(True)
count = await self.dispatcher.skip_updates()
if count:
log.warning(f"Skipped {count} updates.")
return count
async def _welcome(self):
user = await self.dispatcher.bot.me
log.info(f"Bot: {user.full_name} [@{user.username}]")
async def _shutdown(self):
self.dispatcher.stop_polling()
await self.dispatcher.storage.close()
await self.dispatcher.storage.wait_closed()
await self.dispatcher.bot.close()
async def _startup_polling(self):
await self._welcome()
if self.skip_updates:
await self._skip_updates()
for callback in self._on_startup_polling:
await callback(self.dispatcher)
async def _shutdown_polling(self, wait_closed=False):
await self._shutdown()
for callback in self._on_shutdown_polling:
await callback(self.dispatcher)
if wait_closed:
await self.dispatcher.wait_closed()
async def _shutdown_webhook(self, wait_closed=False):
for callback in self._on_shutdown_webhook:
await callback(self.dispatcher)
await self._shutdown()
if wait_closed:
await self.dispatcher.wait_closed()
async def _startup_webhook(self):
await self._welcome()
if self.skip_updates:
self._skip_updates()

16
aiogram/utils/safe.py
View file

@ -1,16 +0,0 @@
import typing
async def safe(coro: typing.Coroutine) -> (bool, typing.Any):
"""
Safety execute coroutine
Status - returns True if success otherwise False
:param coro:
:return: status and result
"""
try:
return True, await coro
except Exception as e:
return False, e

140
aiogram/utils/versions.py
View file

@ -1,140 +0,0 @@
import datetime
import os
import subprocess
from .helper import Helper, HelperMode, Item
# Based on https://github.com/django/django/blob/master/django/utils/version.py
class Version:
def __init__(self, major=0, minor=0,
maintenance=0, stage='final', build=0):
self.__raw_version = None
self.__version = None
self.version = (major, minor, maintenance, stage, build)
@property
def version(self):
if self.__version is None:
self.__version = self.get_version()
return self.__version
@version.setter
def version(self, version):
if not isinstance(version, (tuple, list)):
raise TypeError(f"`version` must be an instance of tuple/list, not {type(version)}")
self.__raw_version = version
self.__version = None
@property
def major(self):
return self.__raw_version[0]
@property
def minor(self):
return self.__raw_version[1]
@property
def maintenance(self):
return self.__raw_version[2]
@property
def stage(self):
return self.__raw_version[3]
@property
def build(self):
return self.__raw_version[4]
@property
def raw_version(self):
return self.raw_version
@property
def pypi_development_status(self):
if self.stage == Stage.DEV:
status = '2 - Pre-Alpha'
elif self.stage == Stage.ALPHA:
status = '3 - Alpha'
elif self.stage == Stage.BETA:
status = '4 - Beta'
elif self.stage == Stage.FINAL:
status = '5 - Production/Stable'
else:
status = '1 - Planning'
return f"Development Status :: {status}"
def get_version(self):
"""
Returns a PEP 440-compliant version number from VERSION.
:param:
:return:
"""
version = self.__raw_version
# Now build the two parts of the version number:
# app = X.Y[.Z]
# sub = .devN - for pre-alpha releases
# | {a|b|rc}N - for alpha, beta, and rc releases
main = self.get_main_version()
sub = ''
if version[3] == Stage.DEV and version[4] == 0:
git_changeset = self.get_git_changeset()
if git_changeset:
sub = '.dev{0}'.format(git_changeset)
elif version[3] != Stage.FINAL:
mapping = {Stage.ALPHA: 'a', Stage.BETA: 'b',
Stage.RC: 'rc', Stage.DEV: 'dev'}
sub = mapping[version[3]] + str(version[4])
return str(main + sub)
def get_main_version(self):
"""
Returns app version (X.Y[.Z]) from VERSION.
:param:
:return:
"""
version = self.__raw_version
parts = 2 if version[2] == 0 else 3
return '.'.join(str(x) for x in version[:parts])
def get_git_changeset(self):
"""Return a numeric identifier of the latest git changeset.
The result is the UTC timestamp of the changeset in YYYYMMDDHHMMSS format.
This value isn't guaranteed to be unique, but collisions are very unlikely,
so it's sufficient for generating the development version numbers.
"""
repo_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
git_log = subprocess.Popen(
'git log --pretty=format:%ct --quiet -1 HEAD',
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
shell=True, cwd=repo_dir, universal_newlines=True,
)
timestamp = git_log.communicate()[0]
try:
timestamp = datetime.datetime.utcfromtimestamp(int(timestamp))
except ValueError:
return None
return timestamp.strftime('%Y%m%d%H%M%S')
def __str__(self):
return self.version
def __repr__(self):
return '<Version:' + str(self) + '>'
class Stage(Helper):
mode = HelperMode.lowercase
FINAL = Item()
ALPHA = Item()
BETA = Item()
RC = Item()
DEV = Item()

16
dev_requirements.txt
View file

@ -1,12 +1,16 @@
-r requirements.txt
ujson>=1.35
emoji>=0.4.5
pytest>=3.3.0
emoji>=0.5.0
pytest>=3.5.0
pytest-asyncio>=0.8.0
uvloop>=0.9.1
aioredis>=1.0.0
wheel>=0.30.0
aioredis>=1.1.0
wheel>=0.31.0
rethinkdb>=2.3.0
sphinx>=1.6.6
sphinx-rtd-theme>=0.2.4
sphinx>=1.7.3
sphinx-rtd-theme>=0.3.0
aresponses>=1.0.0
tox>=3.0.0
aiosocksy>=0.1
click>=6.7

2
docs/source/conf.py
View file

@ -61,7 +61,7 @@ author = 'Illemius / Alex Root Junior'
# built documents.
#
# The short X.Y version.
version = '{0}.{1}'.format(aiogram.VERSION.major, aiogram.VERSION.minor)
version = aiogram.__version__
# The full version, including alpha/beta/rc tags.
release = aiogram.__version__

72
examples/broadcast_example.py Normal file
View file

@ -0,0 +1,72 @@
import asyncio
import logging
from aiogram import Bot, Dispatcher, types
from aiogram.utils import exceptions, executor
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)
def get_users():
"""
Return users list
In this example returns some random ID's
"""
yield from (61043901, 78238238, 78378343, 98765431, 12345678)
async def send_message(user_id: int, text: str) -> bool:
"""
Safe messages sender
:param user_id:
:param text:
:return:
"""
try:
await bot.send_message(user_id, '<b>Hello, World!</b>')
except exceptions.BotBlocked:
log.error(f"Target [ID:{user_id}]: blocked by user")
except exceptions.ChatNotFound:
log.error(f"Target [ID:{user_id}]: invalid user ID")
except exceptions.RetryAfter as e:
log.error(f"Target [ID:{user_id}]: Flood limit is exceeded. Sleep {e.timeout} seconds.")
await asyncio.sleep(e.timeout)
return await send_message(user_id, text) # Recursive call
except exceptions.TelegramAPIError:
log.exception(f"Target [ID:{user_id}]: failed")
else:
log.info(f"Target [ID:{user_id}]: success")
return True
return False
async def broadcaster() -> int:
"""
Simple broadcaster
:return: Count of messages
"""
count = 0
try:
for user_id in get_users():
if await send_message(user_id, '<b>Hello!</b>'):
count += 1
await asyncio.sleep(.05) # 20 messages per second (Limit: 30 messages per second)
finally:
log.info(f"{count} messages successful sent.")
return count
if __name__ == '__main__':
# Execute broadcaster
executor.start(dp, broadcaster())

29
examples/proxy_and_emojize.py
View file

@ -1,6 +1,8 @@
import asyncio
import logging
import aiohttp
from aiogram import Bot, types
from aiogram.dispatcher import Dispatcher
from aiogram.types import ParseMode
@ -10,12 +12,13 @@ from aiogram.utils.markdown import bold, code, italic, text
# Configure bot here
API_TOKEN = 'BOT TOKEN HERE'
PROXY_URL = 'http://PROXY_URL'
PROXY_URL = 'http://PROXY_URL' # Or 'socks5://...'
# 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)
# Also you can use Socks5 proxy but you need manually install aiosocksy package.
# Get my ip URL
GET_IP_URL = 'http://bot.whatismyipaddress.com/'
@ -27,29 +30,29 @@ bot = Bot(token=API_TOKEN, loop=loop, proxy=PROXY_URL)
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()
@dp.message_handler(commands=['start'])
async def cmd_start(message: types.Message):
# Create a temporary session
session = bot.create_temp_session()
content = []
# Make request (without proxy)
async with session.get(GET_IP_URL) as response:
content.append(text(':globe_showing_Americas:', bold('IP:'), code(await response.text())))
# This line is formatted to '🌎 *IP:* `YOUR IP`'
ip = await fetch(GET_IP_URL)
content.append(text(':globe_showing_Americas:', bold('IP:'), code(ip)))
# This line is formatted to '🌎 *IP:* `YOUR IP`'
# Make request through proxy
async with session.get(GET_IP_URL, proxy=bot.proxy, proxy_auth=bot.proxy_auth) as response:
content.append(text(':locked_with_key:', bold('IP:'), code(await response.text()), italic('via proxy')))
# This line is formatted to '🔐 *IP:* `YOUR IP` _via 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')))
# 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)
# Destroy temp session
bot.destroy_temp_session(session)
# 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/
# For representing emoji codes into real emoji use emoji util (aiogram.utils.emoji)

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.

5
requirements.txt
View file

@ -1,2 +1,3 @@
aiohttp>=2.3.5
Babel>=2.5.1
aiohttp>=3.1.3
Babel>=2.5.3
certifi>=2018.4.16

56
setup.py
View file

@ -1,18 +1,34 @@
#!/usr/bin/env python3
import pathlib
import re
import sys
from distutils.core import setup
from warnings import warn
from pip.req import parse_requirements
from setuptools import PackageFinder
from setuptools import find_packages, setup
from aiogram import Stage, VERSION
try:
from pip.req import parse_requirements
except ImportError: # pip >= 10.0.0
from pip._internal.req import parse_requirements
WORK_DIR = pathlib.Path(__file__).parent
# Check python version
MINIMAL_PY_VERSION = (3, 6)
if sys.version_info < MINIMAL_PY_VERSION:
warn('aiogram works only with Python {}+'.format('.'.join(map(str, MINIMAL_PY_VERSION)), RuntimeWarning))
raise RuntimeError('aiogram works only with Python {}+'.format('.'.join(map(str, MINIMAL_PY_VERSION))))
def get_version():
"""
Read version
:return: str
"""
txt = (WORK_DIR / 'aiogram' / '__init__.py').read_text('utf-8')
try:
return re.findall(r"^__version__ = '([^']+)'\r?$", txt, re.M)[0]
except IndexError:
raise RuntimeError('Unable to determine version.')
def get_description():
@ -26,35 +42,35 @@ def get_description():
return f.read()
def get_requirements():
def get_requirements(filename=None):
"""
Read requirements from 'requirements txt'
:return: requirements
:rtype: list
"""
filename = 'requirements.txt'
if VERSION.stage == Stage.DEV:
filename = 'dev_' + filename
if filename is None:
filename = 'requirements.txt'
install_reqs = parse_requirements(filename, session='hack')
file = WORK_DIR / filename
install_reqs = parse_requirements(str(file), session='hack')
return [str(ir.req) for ir in install_reqs]
install_requires = get_requirements()
setup(
name='aiogram',
version=VERSION.version,
packages=PackageFinder.find(exclude=('tests', 'tests.*', 'examples.*', 'docs',)),
version=get_version(),
packages=find_packages(exclude=('tests', 'tests.*', 'examples.*', 'docs',)),
url='https://github.com/aiogram/aiogram',
license='MIT',
author='Alex Root Junior',
author_email='jroot.junior@gmail.com',
requires_python='>=3.6',
author_email='aiogram@illemius.xyz',
description='Is a pretty simple and fully asynchronous library for Telegram Bot API',
long_description=get_description(),
classifiers=[
VERSION.pypi_development_status, # Automated change classifier by build stage
'Development Status :: 5 - Production/Stable',
'Environment :: Console',
'Framework :: AsyncIO',
'Intended Audience :: Developers',
@ -63,5 +79,5 @@ setup(
'Programming Language :: Python :: 3.6',
'Topic :: Software Development :: Libraries :: Application Frameworks',
],
install_requires=install_requires
install_requires=get_requirements()
)

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

41
tests/test_token.py Normal file
View file

@ -0,0 +1,41 @@
import pytest
from aiogram.bot import api
from aiogram.utils import auth_widget, exceptions
VALID_TOKEN = '123456789:AABBCCDDEEFFaabbccddeeff-1234567890'
INVALID_TOKEN = '123456789:AABBCCDDEEFFaabbccddeeff 123456789' # Space in token and wrong length
VALID_DATA = {
'date': 1525385236,
'first_name': 'Test',
'last_name': 'User',
'id': 123456789,
'username': 'username',
'hash': '69a9871558fbbe4cd0dbaba52fa1cc4f38315d3245b7504381a64139fb024b5b'
}
INVALID_DATA = {
'date': 1525385237,
'first_name': 'Test',
'last_name': 'User',
'id': 123456789,
'username': 'username',
'hash': '69a9871558fbbe4cd0dbaba52fa1cc4f38315d3245b7504381a64139fb024b5b'
}
def test_valid_token():
assert api.check_token(VALID_TOKEN)
def test_invalid_token():
with pytest.raises(exceptions.ValidationError):
api.check_token(INVALID_TOKEN)
def test_widget():
assert auth_widget.check_token(VALID_DATA, VALID_TOKEN)
def test_invalid_widget_data():
assert not auth_widget.check_token(INVALID_DATA, VALID_TOKEN)

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
}