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

Refactor types.

Browse source
This commit is contained in:
Alex Root Junior 2018-01-24 02:29:48 +02:00
parent f7ef84acbc
commit 8490306096
21 changed files with 889 additions and 210 deletions
Download patch file Download diff file Expand all files Collapse all files

2
aiogram/bot/api.py
View file

@ -115,7 +115,7 @@ def _compose_data(params=None, files=None):
else:
raise ValueError('Tuple must have exactly 2 elements: filename, fileobj')
elif isinstance(f, types.InputFile):
filename, fileobj = f.get_filename(), f.get_file()
filename, fileobj = f.filename, f.file
else:
filename, fileobj = _guess_filename(f) or key, f

11
aiogram/types/animation.py
View file

@ -1,9 +1,10 @@
from . import base
from . import fields
from . import mixins
from .photo_size import PhotoSize
class Animation(base.TelegramObject):
class Animation(base.TelegramObject, mixins.Downloadable):
"""
You can provide an animation for your game so that it looks stylish in chats
(check out Lumberjack for an example).
@ -17,11 +18,3 @@ class Animation(base.TelegramObject):
file_name: base.String = fields.Field()
mime_type: base.String = fields.Field()
file_size: base.Integer = fields.Field()
def __hash__(self):
return self.file_id
def __eq__(self, other):
if isinstance(other, type(self)):
return other.file_id == self.file_id
return self.file_id == other

15
aiogram/types/audio.py
View file

@ -1,8 +1,9 @@
from . import base
from . import fields
from . import mixins
class Audio(base.TelegramObject):
class Audio(base.TelegramObject, mixins.Downloadable):
"""
This object represents an audio file to be treated as music by the Telegram clients.
@ -16,9 +17,9 @@ class Audio(base.TelegramObject):
file_size: base.Integer = fields.Field()
def __hash__(self):
return self.file_id
def __eq__(self, other):
if isinstance(other, type(self)):
return other.file_id == self.file_id
return self.file_id == other
return hash(self.file_id) + \
self.duration + \
hash(self.performer) + \
hash(self.title) + \
hash(self.mime_type) + \
self.file_size

4
aiogram/types/base.py
View file

@ -5,12 +5,12 @@ from typing import TypeVar
from .fields import BaseField
from ..utils import json
__all__ = ('MetaTelegramObject', 'TelegramObject', 'InputFile', 'String', 'Integer', 'Float', 'Boolean')
PROPS_ATTR_NAME = '_props'
VALUES_ATTR_NAME = '_values'
ALIASES_ATTR_NAME = '_aliases'
__all__ = ('MetaTelegramObject', 'TelegramObject')
# Binding of builtin types
InputFile = TypeVar('InputFile', 'InputFile', io.BytesIO, io.FileIO, str)
String = TypeVar('String', bound=str)

35
aiogram/types/callback_query.py
View file

@ -1,3 +1,5 @@
import typing
from . import base
from . import fields
from .message import Message
@ -26,10 +28,31 @@ class CallbackQuery(base.TelegramObject):
data: base.String = fields.Field()
game_short_name: base.String = fields.Field()
def __hash__(self):
return self.id
async def answer(self, 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):
"""
Use this method to send answers to callback queries sent from inline keyboards.
The answer will be displayed to the user as a notification at the top of the chat screen or as an alert.
def __eq__(self, other):
if isinstance(other, type(self)):
return other.id == self.id
return self.id == other
Alternatively, the user can be redirected to the specified Game URL.
For this option to work, you must first create a game for your bot via @Botfather and accept the terms.
Otherwise, you may use links like t.me/your_bot?start=XXXX that open your bot with a parameter.
Source: https://core.telegram.org/bots/api#answercallbackquery
:param text: Text of the notification. If not specified, nothing will be shown to the user, 0-200 characters
:type text: :obj:`typing.Union[base.String, None]`
:param show_alert: If true, an alert will be shown by the client instead of a notification
at the top of the chat screen. Defaults to false.
:type show_alert: :obj:`typing.Union[base.Boolean, None]`
:param url: URL that will be opened by the user's client.
:type url: :obj:`typing.Union[base.String, None]`
:param cache_time: The maximum amount of time in seconds that the
result of the callback query may be cached client-side.
:type cache_time: :obj:`typing.Union[base.Integer, None]`
:return: On success, True is returned.
:rtype: :obj:`base.Boolean`"""
await self.bot.answer_callback_query(callback_query_id=self.id, text=text,
show_alert=show_alert, url=url, cache_time=cache_time)

271
aiogram/types/chat.py
View file

@ -1,4 +1,5 @@
import asyncio
import typing
from . import base
from . import fields
@ -62,45 +63,303 @@ class Chat(base.TelegramObject):
return markdown.link(name, self.user_url)
async def set_photo(self, photo):
"""
Use this method to set a new profile photo for the chat. Photos can't be changed for private chats.
The bot must be an administrator in the chat for this to work and must have the appropriate admin rights.
Note: In regular groups (non-supergroups), this method will only work if the ‘All Members Are Admins’
setting is off in the target group.
Source: https://core.telegram.org/bots/api#setchatphoto
:param photo: New chat photo, uploaded using multipart/form-data
:type photo: :obj:`base.InputFile`
:return: Returns True on success.
:rtype: :obj:`base.Boolean`
"""
return await self.bot.set_chat_photo(self.id, photo)
async def delete_photo(self):
"""
Use this method to delete a chat photo. Photos can't be changed for private chats.
The bot must be an administrator in the chat for this to work and must have the appropriate admin rights.
Note: In regular groups (non-supergroups), this method will only work if the ‘All Members Are Admins’
setting is off in the target group.
Source: https://core.telegram.org/bots/api#deletechatphoto
:return: Returns True on success.
:rtype: :obj:`base.Boolean`
"""
return await self.bot.delete_chat_photo(self.id)
async def set_title(self, title):
"""
Use this method to change the title of a chat. Titles can't be changed for private chats.
The bot must be an administrator in the chat for this to work and must have the appropriate admin rights.
Note: In regular groups (non-supergroups), this method will only work if the ‘All Members Are Admins’
setting is off in the target group.
Source: https://core.telegram.org/bots/api#setchattitle
:param title: New chat title, 1-255 characters
:type title: :obj:`base.String`
:return: Returns True on success.
:rtype: :obj:`base.Boolean`
"""
return await self.bot.set_chat_title(self.id, title)
async def set_description(self, description):
"""
Use this method to change the description of a supergroup or a channel.
The bot must be an administrator in the chat for this to work and must have the appropriate admin rights.
Source: https://core.telegram.org/bots/api#setchatdescription
:param description: New chat description, 0-255 characters
:type description: :obj:`typing.Union[base.String, None]`
:return: Returns True on success.
:rtype: :obj:`base.Boolean`
"""
return await self.bot.delete_chat_description(self.id, description)
async def kick(self, user_id: base.Integer,
until_date: typing.Union[base.Integer, None] = None):
"""
Use this method to kick a user from a group, a supergroup or a channel.
In the case of supergroups and channels, the user will not be able to return to the group
on their own using invite links, etc., unless unbanned first.
The bot must be an administrator in the chat for this to work and must have the appropriate admin rights.
Note: In regular groups (non-supergroups), this method will only work if the ‘All Members Are Admins’ setting
is off in the target group.
Otherwise members may only be removed by the group's creator or by the member that added them.
Source: https://core.telegram.org/bots/api#kickchatmember
:param user_id: Unique identifier of the target user
:type user_id: :obj:`base.Integer`
:param until_date: Date when the user will be unbanned, unix time.
:type until_date: :obj:`typing.Union[base.Integer, None]`
:return: Returns True on success.
:rtype: :obj:`base.Boolean`
"""
return await self.bot.kick_chat_member(self.id, user_id=user_id, until_date=until_date)
async def unban(self, user_id: base.Integer):
"""
Use this method to unban a previously kicked user in a supergroup or channel. `
The user will not return to the group or channel automatically, but will be able to join via link, etc.
The bot must be an administrator for this to work.
Source: https://core.telegram.org/bots/api#unbanchatmember
:param user_id: Unique identifier of the target user
:type user_id: :obj:`base.Integer`
:return: Returns True on success.
:rtype: :obj:`base.Boolean`
"""
return await self.bot.unban_chat_member(self.id, user_id=user_id)
async def restrict(self, user_id: base.Integer,
until_date: typing.Union[base.Integer, None] = None,
can_send_messages: typing.Union[base.Boolean, None] = None,
can_send_media_messages: typing.Union[base.Boolean, None] = None,
can_send_other_messages: typing.Union[base.Boolean, None] = None,
can_add_web_page_previews: typing.Union[base.Boolean, None] = None) -> base.Boolean:
"""
Use this method to restrict a user in a supergroup.
The bot must be an administrator in the supergroup for this to work and must have the appropriate admin rights.
Pass True for all boolean parameters to lift restrictions from a user.
Source: https://core.telegram.org/bots/api#restrictchatmember
:param user_id: Unique identifier of the target user
:type user_id: :obj:`base.Integer`
:param until_date: Date when restrictions will be lifted for the user, unix time.
:type until_date: :obj:`typing.Union[base.Integer, None]`
:param can_send_messages: Pass True, if the user can send text messages, contacts, locations and venues
:type can_send_messages: :obj:`typing.Union[base.Boolean, None]`
:param can_send_media_messages: Pass True, if the user can send audios, documents, photos, videos,
video notes and voice notes, implies can_send_messages
:type can_send_media_messages: :obj:`typing.Union[base.Boolean, None]`
:param can_send_other_messages: Pass True, if the user can send animations, games, stickers and
use inline bots, implies can_send_media_messages
:type can_send_other_messages: :obj:`typing.Union[base.Boolean, None]`
:param can_add_web_page_previews: Pass True, if the user may add web page previews to their messages,
implies can_send_media_messages
:type can_add_web_page_previews: :obj:`typing.Union[base.Boolean, None]`
:return: Returns True on success.
:rtype: :obj:`base.Boolean`
"""
return self.bot.restrict_chat_member(self.id, user_id=user_id, until_date=until_date,
can_send_messages=can_send_messages,
can_send_media_messages=can_send_media_messages,
can_send_other_messages=can_send_other_messages,
can_add_web_page_previews=can_add_web_page_previews)
async def promote(self, user_id: base.Integer,
can_change_info: typing.Union[base.Boolean, None] = None,
can_post_messages: typing.Union[base.Boolean, None] = None,
can_edit_messages: typing.Union[base.Boolean, None] = None,
can_delete_messages: typing.Union[base.Boolean, None] = None,
can_invite_users: typing.Union[base.Boolean, None] = None,
can_restrict_members: typing.Union[base.Boolean, None] = None,
can_pin_messages: typing.Union[base.Boolean, None] = None,
can_promote_members: typing.Union[base.Boolean, None] = None) -> base.Boolean:
"""
Use this method to promote or demote a user in a supergroup or a channel.
The bot must be an administrator in the chat for this to work and must have the appropriate admin rights.
Pass False for all boolean parameters to demote a user.
Source: https://core.telegram.org/bots/api#promotechatmember
:param user_id: Unique identifier of the target user
:type user_id: :obj:`base.Integer`
:param can_change_info: Pass True, if the administrator can change chat title, photo and other settings
:type can_change_info: :obj:`typing.Union[base.Boolean, None]`
:param can_post_messages: Pass True, if the administrator can create channel posts, channels only
:type can_post_messages: :obj:`typing.Union[base.Boolean, None]`
:param can_edit_messages: Pass True, if the administrator can edit messages of other users, channels only
:type can_edit_messages: :obj:`typing.Union[base.Boolean, None]`
:param can_delete_messages: Pass True, if the administrator can delete messages of other users
:type can_delete_messages: :obj:`typing.Union[base.Boolean, None]`
:param can_invite_users: Pass True, if the administrator can invite new users to the chat
:type can_invite_users: :obj:`typing.Union[base.Boolean, None]`
:param can_restrict_members: Pass True, if the administrator can restrict, ban or unban chat members
:type can_restrict_members: :obj:`typing.Union[base.Boolean, None]`
:param can_pin_messages: Pass True, if the administrator can pin messages, supergroups only
:type can_pin_messages: :obj:`typing.Union[base.Boolean, None]`
:param can_promote_members: Pass True, if the administrator can add new administrators
with a subset of his own privileges or demote administrators that he has promoted,
directly or indirectly (promoted by administrators that were appointed by him)
:type can_promote_members: :obj:`typing.Union[base.Boolean, None]`
:return: Returns True on success.
:rtype: :obj:`base.Boolean`
"""
return self.bot.promote_chat_member(self.id,
can_change_info=can_change_info,
can_post_messages=can_post_messages,
can_edit_messages=can_edit_messages,
can_delete_messages=can_delete_messages,
can_invite_users=can_invite_users,
can_restrict_members=can_restrict_members,
can_pin_messages=can_pin_messages,
can_promote_members=can_promote_members)
async def pin_message(self, message_id: int, disable_notification: bool = False):
"""
Use this method to pin a message in a supergroup.
The bot must be an administrator in the chat for this to work and must have the appropriate admin rights.
Source: https://core.telegram.org/bots/api#pinchatmessage
:param message_id: Identifier of a message to pin
:type message_id: :obj:`base.Integer`
:param disable_notification: Pass True, if it is not necessary to send a notification to
all group members about the new pinned message
:type disable_notification: :obj:`typing.Union[base.Boolean, None]`
:return: Returns True on success.
:rtype: :obj:`base.Boolean`
"""
return await self.bot.pin_chat_message(self.id, message_id, disable_notification)
async def unpin_message(self):
"""
Use this method to unpin a message in a supergroup chat.
The bot must be an administrator in the chat for this to work and must have the appropriate admin rights.
Source: https://core.telegram.org/bots/api#unpinchatmessage
:return: Returns True on success.
:rtype: :obj:`base.Boolean`
"""
return await self.bot.unpin_chat_message(self.id)
async def leave(self):
"""
Use this method for your bot to leave a group, supergroup or channel.
Source: https://core.telegram.org/bots/api#leavechat
:return: Returns True on success.
:rtype: :obj:`base.Boolean`
"""
return await self.bot.leave_chat(self.id)
async def get_administrators(self):
"""
Use this method to get a list of administrators in a chat.
Source: https://core.telegram.org/bots/api#getchatadministrators
:return: On success, returns an Array of ChatMember objects that contains information about all
chat administrators except other bots.
If the chat is a group or a supergroup and no administrators were appointed,
only the creator will be returned.
:rtype: :obj:`typing.List[types.ChatMember]`
"""
return await self.bot.get_chat_administrators(self.id)
async def get_members_count(self):
"""
Use this method to get the number of members in a chat.
Source: https://core.telegram.org/bots/api#getchatmemberscount
:return: Returns Int on success.
:rtype: :obj:`base.Integer`
"""
return await self.bot.get_chat_members_count(self.id)
async def get_member(self, user_id):
"""
Use this method to get information about a member of a chat.
Source: https://core.telegram.org/bots/api#getchatmember
:param user_id: Unique identifier of the target user
:type user_id: :obj:`base.Integer`
:return: Returns a ChatMember object on success.
:rtype: :obj:`types.ChatMember`
"""
return await self.bot.get_chat_member(self.id, user_id)
async def do(self, action):
"""
Use this method when you need to tell the user that something is happening on the bot's side.
The status is set for 5 seconds or less
(when a message arrives from your bot, Telegram clients clear its typing status).
We only recommend using this method when a response from the bot will take
a noticeable amount of time to arrive.
Source: https://core.telegram.org/bots/api#sendchataction
:param action: Type of action to broadcast.
:type action: :obj:`base.String`
:return: Returns True on success.
:rtype: :obj:`base.Boolean`
"""
return await self.bot.send_chat_action(self.id, action)
def __hash__(self):
return self.id
async def export_invite_link(self):
"""
Use this method to export an invite link to a supergroup or a channel.
The bot must be an administrator in the chat for this to work and must have the appropriate admin rights.
def __eq__(self, other):
if isinstance(other, type(self)):
return other.id == self.id
return self.id == other
Source: https://core.telegram.org/bots/api#exportchatinvitelink
:return: Returns exported invite link as String on success.
:rtype: :obj:`base.String`
"""
if self.invite_link:
return self.invite_link
return await self.bot.export_chat_invite_link(self.id)
def __int__(self):
return self.id

10
aiogram/types/chat_member.py
View file

@ -29,13 +29,11 @@ class ChatMember(base.TelegramObject):
can_send_other_messages: base.Boolean = fields.Field()
can_add_web_page_previews: base.Boolean = fields.Field()
def __hash__(self):
return self.user.id
def is_admin(self):
return ChatMemberStatus.is_admin(self.status)
def __eq__(self, other):
if isinstance(other, type(self)):
return other.user.id == self.user.id
return self.user.id == other
def is_member(self):
return ChatMemberStatus.is_member(self.status)
def __int__(self):
return self.user.id

63
aiogram/types/chat_photo.py
View file

@ -1,3 +1,6 @@
import os
import pathlib
from . import base
from . import fields
@ -10,3 +13,63 @@ class ChatPhoto(base.TelegramObject):
"""
small_file_id: base.String = fields.Field()
big_file_id: base.String = fields.Field()
async def download_small(self, destination=None, timeout=30, chunk_size=65536, seek=True, make_dirs=True):
"""
Download file
:param destination: filename or instance of :class:`io.IOBase`. For e. g. :class:`io.BytesIO`
:param timeout: Integer
:param chunk_size: Integer
:param seek: Boolean - go to start of file when downloading is finished.
:param make_dirs: Make dirs if not exist
:return: destination
"""
file = await self.get_small_file()
is_path = True
if destination is None:
destination = file.file_path
elif isinstance(destination, (str, pathlib.Path)) and os.path.isdir(destination):
os.path.join(destination, file.file_path)
else:
is_path = False
if is_path and make_dirs:
os.makedirs(os.path.dirname(destination), exist_ok=True)
return await self.bot.download_file(file_path=file.file_path, destination=destination, timeout=timeout,
chunk_size=chunk_size, seek=seek)
async def download_big(self, destination=None, timeout=30, chunk_size=65536, seek=True, make_dirs=True):
"""
Download file
:param destination: filename or instance of :class:`io.IOBase`. For e. g. :class:`io.BytesIO`
:param timeout: Integer
:param chunk_size: Integer
:param seek: Boolean - go to start of file when downloading is finished.
:param make_dirs: Make dirs if not exist
:return: destination
"""
file = await self.get_big_file()
is_path = True
if destination is None:
destination = file.file_path
elif isinstance(destination, (str, pathlib.Path)) and os.path.isdir(destination):
os.path.join(destination, file.file_path)
else:
is_path = False
if is_path and make_dirs:
os.makedirs(os.path.dirname(destination), exist_ok=True)
return await self.bot.download_file(file_path=file.file_path, destination=destination, timeout=timeout,
chunk_size=chunk_size, seek=seek)
async def get_small_file(self):
return await self.bot.get_file(self.small_file_id)
async def get_big_file(self):
return await self.bot.get_file(self.big_file_id)

7
aiogram/types/contact.py
View file

@ -12,3 +12,10 @@ class Contact(base.TelegramObject):
first_name: base.String = fields.Field()
last_name: base.String = fields.Field()
user_id: base.Integer = fields.Field()
@property
def full_name(self):
name = self.first_name
if self.last_name is not None:
name += ' ' + self.last_name
return name

11
aiogram/types/document.py
View file

@ -1,9 +1,10 @@
from . import base
from . import fields
from . import mixins
from .photo_size import PhotoSize
class Document(base.TelegramObject):
class Document(base.TelegramObject, mixins.Downloadable):
"""
This object represents a general file (as opposed to photos, voice messages and audio files).
@ -14,11 +15,3 @@ class Document(base.TelegramObject):
file_name: base.String = fields.Field()
mime_type: base.String = fields.Field()
file_size: base.Integer = fields.Field()
def __hash__(self):
return self.file_id
def __eq__(self, other):
if isinstance(other, type(self)):
return other.file_id == self.file_id
return self.file_id == other

23
aiogram/types/file.py
View file

@ -1,8 +1,9 @@
from . import base
from . import fields
from . import mixins
class File(base.TelegramObject):
class File(base.TelegramObject, mixins.Downloadable):
"""
This object represents a file ready to be downloaded.
@ -18,23 +19,3 @@ class File(base.TelegramObject):
file_id: base.String = fields.Field()
file_size: base.Integer = fields.Field()
file_path: base.String = fields.Field()
async def download(self, destination=None, timeout=30, chunk_size=65536, seek=True):
"""
Download file by file_path to destination
:param destination: filename or instance of :class:`io.IOBase`. For e. g. :class:`io.BytesIO`
:param timeout: Integer
:param chunk_size: Integer
:param seek: Boolean - go to start of file when downloading is finished.
:return: destination
"""
return await self.bot.download_file(self.file_path, destination, timeout, chunk_size, seek)
def __hash__(self):
return self.file_id
def __eq__(self, other):
if isinstance(other, type(self)):
return other.file_id == self.file_id
return self.file_id == other

13
aiogram/types/inline_keyboard.py
View file

@ -54,10 +54,21 @@ class InlineKeyboardMarkup(base.TelegramObject):
"""
btn_array = []
for button in args:
btn_array.append(button.to_python())
btn_array.append(button)
self.inline_keyboard.append(btn_array)
return self
def insert(self, button):
"""
Insert button to last row
:param button:
"""
if self.inline_keyboard and len(self.inline_keyboard[-1] < self.row_width):
self.inline_keyboard[-1].append(button)
else:
self.add(button)
class InlineKeyboardButton(base.TelegramObject):
"""

8
aiogram/types/inline_query.py
View file

@ -17,11 +17,3 @@ class InlineQuery(base.TelegramObject):
location: Location = fields.Field(base=Location)
query: base.String = fields.Field()
offset: base.String = fields.Field()
def __hash__(self):
return self.id
def __eq__(self, other):
if isinstance(other, type(self)):
return other.id == self.id
return self.id == other

102
aiogram/types/input_file.py
View file

@ -1,10 +1,6 @@
import io
import logging
import os
import tempfile
import time
import aiohttp
from . import base
from ..bot import api
@ -36,11 +32,11 @@ class InputFile(base.TelegramObject):
self._path = path_or_bytesio
if filename is None:
filename = os.path.split(path_or_bytesio)[-1]
else:
# As io.BytesIO
assert isinstance(path_or_bytesio, io.IOBase)
elif isinstance(path_or_bytesio, io.IOBase):
self._path = None
self._file = path_or_bytesio
else:
raise TypeError('Not supported file type.')
self._filename = filename
@ -48,14 +44,17 @@ class InputFile(base.TelegramObject):
"""
Close file descriptor
"""
if not hasattr(self, '_file'):
return
self._file.close()
del self._file
if self.conf.get('downloaded') and self.conf.get('temp'):
log.debug(f"Unlink file '{self._path}'")
os.unlink(self._path)
@property
def filename(self):
if self._filename is None:
self._filename = api._guess_filename(self._file)
return self._filename
@filename.setter
def filename(self, value):
self._filename = value
def get_filename(self) -> str:
"""
@ -63,9 +62,11 @@ class InputFile(base.TelegramObject):
:return: name
"""
if self._filename is None:
self._filename = api._guess_filename(self._file)
return self._filename
return self.filename
@property
def file(self):
return self._file
def get_file(self):
"""
@ -73,74 +74,7 @@ class InputFile(base.TelegramObject):
:return:
"""
return self._file
@classmethod
async def from_url(cls, url, filename=None, temp_file=False, chunk_size=65536):
"""
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 temp_file: use temporary file
: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:
if temp_file:
# Create temp file
fd, path = tempfile.mkstemp(suffix=file_suffix, prefix=filename_prefix + '_')
file = conf['temp'] = path
# Save file in temp directory
with open(fd, 'wb') as f:
await cls._process_stream(response, f, chunk_size=chunk_size)
else:
# 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)
@classmethod
async def _process_stream(cls, response, writer, chunk_size=65536):
"""
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
return self.file
def to_python(self):
raise TypeError('Object of this type is not exportable!')

11
aiogram/types/input_media.py
View file

@ -30,13 +30,16 @@ class InputMedia(base.TelegramObject):
@file.setter
def file(self, file: io.IOBase):
setattr(self, '_file', file)
self.media = ATTACHMENT_PREFIX + secrets.token_urlsafe(16)
attachment_key = self.attachment_key = secrets.token_urlsafe(16)
self.media = ATTACHMENT_PREFIX + attachment_key
@property
def attachment_key(self):
if self.media.startswith(ATTACHMENT_PREFIX):
return self.media[len(ATTACHMENT_PREFIX):]
return None
return self.conf.get('attachment_key', None)
@attachment_key.setter
def attachment_key(self, value):
self.conf['attachment_key'] = value
class InputMediaPhoto(InputMedia):

403
aiogram/types/message.py
View file

@ -21,7 +21,6 @@ from .video import Video
from .video_note import VideoNote
from .voice import Voice
from ..utils import helper
from ..utils.payload import generate_payload
class Message(base.TelegramObject):
@ -177,7 +176,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) -> 'Message':
disable_notification=None, reply_markup=None, reply=False) -> 'Message':
"""
Reply to this message
@ -186,10 +185,360 @@ class Message(base.TelegramObject):
:param disable_web_page_preview: bool
:param disable_notification: bool
:param reply_markup:
:return: :class:`aoigram.types.Message`
:param reply: fill 'reply_to_message_id'
:return: :class:`aiogram.types.Message`
"""
return await self.bot.send_message(self.chat.id, text, parse_mode, disable_web_page_preview,
disable_notification, self.message_id, reply_markup)
return await self.bot.send_message(self.chat.id, text,
parse_mode=parse_mode,
disable_web_page_preview=disable_web_page_preview,
disable_notification=disable_notification,
reply_to_message_id=self.message_id if reply else None,
reply_markup=reply_markup)
async def reply_photo(self, photo: typing.Union[base.InputFile, base.String],
caption: typing.Union[base.String, None] = None,
disable_notification: typing.Union[base.Boolean, None] = None,
reply_markup=None, reply=True) -> 'Message':
"""
Use this method to send photos.
Source: https://core.telegram.org/bots/api#sendphoto
:param photo: Photo to send.
: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 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_photo(self.chat.id, photo=photo, caption=caption,
disable_notification=disable_notification,
reply_to_message_id=self.message_id if reply else None,
reply_markup=reply_markup)
async def reply_audio(self, audio: typing.Union[base.InputFile, base.String],
caption: 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,
disable_notification: typing.Union[base.Boolean, None] = None,
reply_markup=None,
reply=True) -> 'Message':
"""
Use this method to send audio files, if you want Telegram clients to display them in the music player.
Your audio must be in the .mp3 format.
For sending voice messages, use the sendVoice method instead.
Source: https://core.telegram.org/bots/api#sendaudio
:param audio: Audio file to send.
: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 duration: Duration of the audio in seconds
:type duration: :obj:`typing.Union[base.Integer, None]`
:param performer: Performer
:type performer: :obj:`typing.Union[base.String, None]`
:param title: Track name
:type title: :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_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_audio(self.chat.id,
audio=audio,
caption=caption,
duration=duration,
performer=performer,
title=title,
disable_notification=disable_notification,
reply_to_message_id=self.message_id if reply else None,
reply_markup=reply_markup)
async def reply_document(self, document: typing.Union[base.InputFile, base.String],
caption: typing.Union[base.String, None] = None,
disable_notification: typing.Union[base.Boolean, None] = None,
reply_markup=None,
reply=True) -> 'Message':
"""
Use this method to send general files.
Bots can currently send files of any type of up to 50 MB in size, this limit may be changed in the future.
Source: https://core.telegram.org/bots/api#senddocument
:param document: File to send.
: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 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_document(self.chat.id,
document=document,
caption=caption,
disable_notification=disable_notification,
reply_to_message_id=self.message_id if reply else None,
reply_markup=reply_markup)
async def reply_video(self, video: typing.Union[base.InputFile, base.String],
duration: typing.Union[base.Integer, None] = None,
width: typing.Union[base.Integer, None] = None,
height: typing.Union[base.Integer, None] = None,
caption: typing.Union[base.String, None] = None,
disable_notification: typing.Union[base.Boolean, None] = None,
reply_markup=None,
reply=True) -> 'Message':
"""
Use this method to send video files, Telegram clients support mp4 videos
(other formats may be sent as Document).
Source: https://core.telegram.org/bots/api#sendvideo
:param video: Video to send.
:type video: :obj:`typing.Union[base.InputFile, base.String]`
:param duration: Duration of sent video in seconds
:type duration: :obj:`typing.Union[base.Integer, None]`
:param width: Video width
:type width: :obj:`typing.Union[base.Integer, None]`
:param height: Video height
: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 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_video(self.chat.id,
video=video,
duration=duration,
width=width,
height=height,
caption=caption,
disable_notification=disable_notification,
reply_to_message_id=self.message_id if reply else None,
reply_markup=reply_markup)
async def reply_voice(self, voice: typing.Union[base.InputFile, base.String],
caption: typing.Union[base.String, None] = None,
duration: typing.Union[base.Integer, None] = None,
disable_notification: typing.Union[base.Boolean, None] = None,
reply_markup=None,
reply=True) -> 'Message':
"""
Use this method to send audio files, if you want Telegram clients to display the file
as a playable voice message.
For this to work, your audio must be in an .ogg file encoded with OPUS
(other formats may be sent as Audio or Document).
Source: https://core.telegram.org/bots/api#sendvoice
:param voice: Audio file to send.
: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 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.
: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_voice(self.chat.id,
voice=voice,
caption=caption,
duration=duration,
disable_notification=disable_notification,
reply_to_message_id=self.message_id if reply else None,
reply_markup=reply_markup)
async def reply_video_note(self, video_note: typing.Union[base.InputFile, base.String],
duration: typing.Union[base.Integer, None] = None,
length: typing.Union[base.Integer, None] = None,
disable_notification: typing.Union[base.Boolean, None] = None,
reply_markup=None,
reply=True) -> 'Message':
"""
As of v.4.0, Telegram clients support rounded square mp4 videos of up to 1 minute long.
Use this method to send video messages.
Source: https://core.telegram.org/bots/api#sendvideonote
:param video_note: Video note to send.
:type video_note: :obj:`typing.Union[base.InputFile, base.String]`
:param duration: Duration of sent video in seconds
:type duration: :obj:`typing.Union[base.Integer, None]`
:param length: Video width and height
:type length: :obj:`typing.Union[base.Integer, 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_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_video_note(self.chat.id,
video_note=video_note,
duration=duration,
length=length,
disable_notification=disable_notification,
reply_to_message_id=self.message_id if reply else None,
reply_markup=reply_markup)
async def reply_media_group(self, media: typing.Union['MediaGroup', typing.List],
disable_notification: typing.Union[base.Boolean, None] = None,
reply=True) -> typing.List['Message']:
"""
Use this method to send a group of photos or videos as an album.
Source: https://core.telegram.org/bots/api#sendmediagroup
:param media: A JSON-serialized array describing photos and videos to be sent
:type media: :obj:`typing.Union[types.MediaGroup, typing.List]`
: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: fill 'reply_to_message_id'
:return: On success, an array of the sent Messages is returned.
:rtype: typing.List[types.Message]
"""
return await self.bot.send_media_group(self.chat.id,
media=media,
disable_notification=disable_notification,
reply_to_message_id=self.message_id if reply else None)
async def reply_location(self, latitude: base.Float,
longitude: base.Float, live_period: typing.Union[base.Integer, None] = None,
disable_notification: typing.Union[base.Boolean, None] = None,
reply_markup=None,
reply=True) -> 'Message':
"""
Use this method to send point on the map.
Source: https://core.telegram.org/bots/api#sendlocation
:param latitude: Latitude of the location
:type latitude: :obj:`base.Float`
:param longitude: Longitude of the location
:type longitude: :obj:`base.Float`
:param live_period: Period in seconds for which the location will be updated
:type live_period: :obj:`typing.Union[base.Integer, 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_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_location(self.chat.id,
latitude=latitude,
longitude=longitude,
live_period=live_period,
disable_notification=disable_notification,
reply_to_message_id=self.message_id if reply else None,
reply_markup=reply_markup)
async def send_venue(self, 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_markup=None,
reply=True) -> 'Message':
"""
Use this method to send information about a venue.
Source: https://core.telegram.org/bots/api#sendvenue
:param latitude: Latitude of the venue
:type latitude: :obj:`base.Float`
:param longitude: Longitude of the venue
:type longitude: :obj:`base.Float`
:param title: Name of the venue
:type title: :obj:`base.String`
:param address: Address of the venue
:type address: :obj:`base.String`
:param foursquare_id: Foursquare identifier of the venue
:type foursquare_id: :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_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_venue(self.chat.id,
latitude=latitude,
longitude=longitude,
title=title,
address=address,
foursquare_id=foursquare_id,
disable_notification=disable_notification,
reply_to_message_id=self.message_id if reply else None,
reply_markup=reply_markup)
async def send_contact(self, 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_markup=None,
reply=True) -> 'Message':
"""
Use this method to send phone contacts.
Source: https://core.telegram.org/bots/api#sendcontact
:param phone_number: Contact's phone number
:type phone_number: :obj:`base.String`
:param first_name: Contact's first name
:type first_name: :obj:`base.String`
:param last_name: Contact's last name
:type last_name: :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_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_contact(self.chat.id,
phone_number=phone_number,
first_name=first_name, last_name=last_name,
disable_notification=disable_notification,
reply_to_message_id=self.message_id if reply else None,
reply_markup=reply_markup)
async def forward(self, chat_id, disable_notification=None) -> 'Message':
"""
@ -204,12 +553,30 @@ class Message(base.TelegramObject):
async def edit_text(self, text: base.String,
parse_mode: typing.Union[base.String, None] = None,
disable_web_page_preview: typing.Union[base.Boolean, None] = None,
reply_markup: typing.Union['types.InlineKeyboardMarkup',
None] = None):
payload = generate_payload(**locals())
payload['message_id'] = self.message_id
payload['chat_id'] = self.chat.id
return await self.bot.edit_message_text(**payload)
reply_markup=None):
"""
Use this method to edit text and game messages sent by the bot or via the bot (for inline bots).
Source: https://core.telegram.org/bots/api#editmessagetext
:param text: New text of the message
:type text: :obj:`base.String`
: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_web_page_preview: Disables link previews for links in this message
:type disable_web_page_preview: :obj:`typing.Union[base.Boolean, 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, otherwise True is returned.
:rtype: :obj:`typing.Union[types.Message, base.Boolean]`
"""
return await self.bot.edit_message_text(text=text,
chat_id=self.chat.id, message_id=self.message_id,
parse_mode=parse_mode,
disable_web_page_preview=disable_web_page_preview,
reply_markup=reply_markup)
async def delete(self):
"""
@ -220,16 +587,14 @@ class Message(base.TelegramObject):
return await self.bot.delete_message(self.chat.id, self.message_id)
async def pin(self, disable_notification: bool = False):
"""
Pin message
:param disable_notification:
:return:
"""
return await self.chat.pin_message(self.message_id, disable_notification)
def __hash__(self):
return self.message_id
def __eq__(self, other):
if isinstance(other, type(self)):
return other.message_id == self.message_id
return self.message_id == other
def __int__(self):
return self.message_id

46
aiogram/types/mixins.py Normal file
View file

@ -0,0 +1,46 @@
import os
import pathlib
class Downloadable:
"""
Mixin for files
"""
async def download(self, destination=None, timeout=30, chunk_size=65536, seek=True, make_dirs=True):
"""
Download file
:param destination: filename or instance of :class:`io.IOBase`. For e. g. :class:`io.BytesIO`
:param timeout: Integer
:param chunk_size: Integer
:param seek: Boolean - go to start of file when downloading is finished.
:param make_dirs: Make dirs if not exist
:return: destination
"""
file = await self.get_file()
is_path = True
if destination is None:
destination = file.file_path
elif isinstance(destination, (str, pathlib.Path)) and os.path.isdir(destination):
os.path.join(destination, file.file_path)
else:
is_path = False
if is_path and make_dirs:
os.makedirs(os.path.dirname(destination), exist_ok=True)
return await self.bot.download_file(file_path=file.file_path, destination=destination, timeout=timeout,
chunk_size=chunk_size, seek=seek)
async def get_file(self):
"""
Get file information
:return: :obj:`aiogram.types.File`
"""
if hasattr(self, 'file_path'):
return self
else:
return await self.bot.get_file(self.file_id)

3
aiogram/types/photo_size.py
View file

@ -1,8 +1,9 @@
from . import base
from . import fields
from . import mixins
class PhotoSize(base.TelegramObject):
class PhotoSize(base.TelegramObject, mixins.Downloadable):
"""
This object represents one size of a photo or a file / sticker thumbnail.

41
aiogram/types/reply_keyboard.py
View file

@ -33,32 +33,45 @@ class ReplyKeyboardMarkup(base.TelegramObject):
self.conf['row_width'] = value
def add(self, *args):
i = 1
"""
Add buttons
:param args:
:return:
"""
row = []
for button in args:
if isinstance(button, str):
row.append({'text': button})
elif isinstance(button, bytes):
row.append({'text': button.decode('utf-8')})
else:
row.append(button.to_python())
if i % self.row_width == 0:
for index, button in enumerate(args):
row.append(button)
if index % self.row_width == 0:
self.keyboard.append(row)
row = []
i += 1
if len(row) > 0:
self.keyboard.append(row)
def row(self, *args):
"""
Add row
:param args:
:return:
"""
btn_array = []
for button in args:
if isinstance(button, str):
btn_array.append({'text': button})
else:
btn_array.append(button.to_python())
btn_array.append(button)
self.keyboard.append(btn_array)
return self
def insert(self, button):
"""
Insert button to last row
:param button:
"""
if self.keyboard and len(self.keyboard[-1] < self.row_width):
self.keyboard[-1].append(button)
else:
self.add(button)
class KeyboardButton(base.TelegramObject):
"""

5
aiogram/types/update.py
View file

@ -30,11 +30,6 @@ class Update(base.TelegramObject):
def __hash__(self):
return self.update_id
def __eq__(self, other):
if isinstance(other, type(self)):
return other.update_id == self.update_id
return self.update_id == other
def __int__(self):
return self.update_id

15
aiogram/types/user.py
View file

@ -64,7 +64,10 @@ class User(base.TelegramObject):
def url(self):
return f"tg://user?id={self.id}"
def get_mention(self, name=None, as_html=False):
def get_mention(self, name=None, as_html=None):
if as_html is None and self.bot.parse_mode and self.bot.parse_mode.lower() == 'html':
as_html = True
if name is None:
name = self.mention
if as_html:
@ -75,12 +78,10 @@ class User(base.TelegramObject):
return await self.bot.get_user_profile_photos(self.id, offset, limit)
def __hash__(self):
return self.id
def __eq__(self, other):
if isinstance(other, type(self)):
return other.id == self.id
return self.id == other
return self.id + \
hash(self.is_bot) + \
hash(self.full_name) + \
(hash(self.username) if self.username else 0)
def __int__(self):
return self.id