import logging
import os
from dataclasses import dataclass
from http import HTTPStatus
import aiohttp
from .. import types
from ..utils import exceptions, json
from ..utils.helper import Helper, HelperMode, Item
# Main aiogram logger
log = logging.getLogger('aiogram')
[docs]@dataclass(frozen=True)
class TelegramAPIServer:
"""
Base config for API Endpoints
"""
base: str
file: str
[docs] def api_url(self, token: str, method: str) -> str:
"""
Generate URL for API methods
:param token: Bot token
:param method: API method name (case insensitive)
:return: URL
"""
return self.base.format(token=token, method=method)
[docs] def file_url(self, token: str, path: str) -> str:
"""
Generate URL for downloading files
:param token: Bot token
:param path: file path
:return: URL
"""
return self.file.format(token=token, path=path)
@classmethod
def from_base(cls, base: str) -> 'TelegramAPIServer':
base = base.rstrip("/")
return cls(
base=f"{base}/bot{{token}}/{{method}}",
file=f"{base}/file/bot{{token}}/{{path}}",
)
TELEGRAM_PRODUCTION = TelegramAPIServer.from_base("https://api.telegram.org")
[docs]def check_token(token: str) -> bool:
"""
Validate BOT token
:param token:
:return:
"""
if not isinstance(token, str):
message = (f"Token is invalid! "
f"It must be 'str' type instead of {type(token)} type.")
raise exceptions.ValidationError(message)
if any(x.isspace() for x in token):
message = "Token is invalid! It can't contains spaces."
raise exceptions.ValidationError(message)
left, sep, right = token.partition(':')
if (not sep) or (not left.isdigit()) or (not right):
raise exceptions.ValidationError('Token is invalid!')
return True
[docs]def check_result(method_name: str, content_type: str, status_code: int, body: str):
"""
Checks whether `result` is a valid API response.
A result is considered invalid if:
- The server returned an HTTP response code other than 200
- The content of the result is invalid JSON.
- The method call was unsuccessful (The JSON 'ok' field equals False)
:param method_name: The name of the method called
:param status_code: status code
:param content_type: content type of result
:param body: result body
:return: The result parsed to a JSON dictionary
:raises ApiException: if one of the above listed cases is applicable
"""
log.debug('Response for %s: [%d] "%r"', method_name, status_code, body)
if content_type != 'application/json':
raise exceptions.NetworkError(f"Invalid response with content type {content_type}: \"{body}\"")
try:
result_json = json.loads(body)
except ValueError:
result_json = {}
description = result_json.get('description') or body
parameters = types.ResponseParameters(**result_json.get('parameters', {}) or {})
if HTTPStatus.OK <= status_code <= HTTPStatus.IM_USED:
return result_json.get('result')
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 status_code == HTTPStatus.BAD_REQUEST:
exceptions.BadRequest.detect(description)
elif status_code == HTTPStatus.NOT_FOUND:
exceptions.NotFound.detect(description)
elif status_code == HTTPStatus.CONFLICT:
exceptions.ConflictError.detect(description)
elif status_code in (HTTPStatus.UNAUTHORIZED, HTTPStatus.FORBIDDEN):
exceptions.Unauthorized.detect(description)
elif status_code == 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 status_code >= HTTPStatus.INTERNAL_SERVER_ERROR:
if 'restart' in description:
raise exceptions.RestartingTelegram()
raise exceptions.TelegramAPIError(description)
raise exceptions.TelegramAPIError(f"{description} [{status_code}]")
async def make_request(session, server, token, method, data=None, files=None, **kwargs):
log.debug('Make request: "%s" with data: "%r" and files "%r"', method, data, files)
url = server.api_url(token=token, method=method)
req = compose_data(data, files)
try:
async with session.post(url, data=req, **kwargs) as response:
return check_result(method, response.content_type, response.status, await response.text())
except aiohttp.ClientError as e:
raise exceptions.NetworkError(f"aiohttp client throws an error: {e.__class__.__name__}: {e}")
[docs]def guess_filename(obj):
"""
Get file name from object
:param obj:
:return:
"""
name = getattr(obj, 'name', None)
if name and isinstance(name, str) and name[0] != '<' and name[-1] != '>':
return os.path.basename(name)
[docs]def compose_data(params=None, files=None):
"""
Prepare request data
:param params:
:param files:
:return:
"""
data = aiohttp.formdata.FormData(quote_fields=False)
if params:
for key, value in params.items():
data.add_field(key, str(value))
if files:
for key, f in files.items():
if isinstance(f, tuple):
if len(f) == 2:
filename, fileobj = f
else:
raise ValueError('Tuple must have exactly 2 elements: filename, fileobj')
elif isinstance(f, types.InputFile):
filename, fileobj = f.filename, f.file
else:
filename, fileobj = guess_filename(f) or key, f
data.add_field(key, fileobj, filename=filename)
return data
[docs]class Methods(Helper):
"""
Helper for Telegram API Methods listed on https://core.telegram.org/bots/api
"""
mode = HelperMode.lowerCamelCase
# Getting Updates
GET_UPDATES = Item() # getUpdates
SET_WEBHOOK = Item() # setWebhook
DELETE_WEBHOOK = Item() # deleteWebhook
GET_WEBHOOK_INFO = Item() # getWebhookInfo
# Available methods
GET_ME = Item() # getMe
LOG_OUT = Item() # logOut
CLOSE = Item() # close
SEND_MESSAGE = Item() # sendMessage
FORWARD_MESSAGE = Item() # forwardMessage
COPY_MESSAGE = Item() # copyMessage
SEND_PHOTO = Item() # sendPhoto
SEND_AUDIO = Item() # sendAudio
SEND_DOCUMENT = Item() # sendDocument
SEND_VIDEO = Item() # sendVideo
SEND_ANIMATION = Item() # sendAnimation
SEND_VOICE = Item() # sendVoice
SEND_VIDEO_NOTE = Item() # sendVideoNote
SEND_MEDIA_GROUP = Item() # sendMediaGroup
SEND_LOCATION = Item() # sendLocation
EDIT_MESSAGE_LIVE_LOCATION = Item() # editMessageLiveLocation
STOP_MESSAGE_LIVE_LOCATION = Item() # stopMessageLiveLocation
SEND_VENUE = Item() # sendVenue
SEND_CONTACT = Item() # sendContact
SEND_POLL = Item() # sendPoll
SEND_DICE = Item() # sendDice
SEND_CHAT_ACTION = Item() # sendChatAction
GET_USER_PROFILE_PHOTOS = Item() # getUserProfilePhotos
GET_FILE = Item() # getFile
KICK_CHAT_MEMBER = Item() # kickChatMember
BAN_CHAT_MEMBER = Item() # banChatMember
UNBAN_CHAT_MEMBER = Item() # unbanChatMember
RESTRICT_CHAT_MEMBER = Item() # restrictChatMember
PROMOTE_CHAT_MEMBER = Item() # promoteChatMember
SET_CHAT_ADMINISTRATOR_CUSTOM_TITLE = Item() # setChatAdministratorCustomTitle
BAN_CHAT_SENDER_CHAT = Item() # banChatSenderChat
UNBAN_CHAT_SENDER_CHAT = Item() # unbanChatSenderChat
SET_CHAT_PERMISSIONS = Item() # setChatPermissions
EXPORT_CHAT_INVITE_LINK = Item() # exportChatInviteLink
CREATE_CHAT_INVITE_LINK = Item() # createChatInviteLink
EDIT_CHAT_INVITE_LINK = Item() # editChatInviteLink
REVOKE_CHAT_INVITE_LINK = Item() # revokeChatInviteLink
APPROVE_CHAT_JOIN_REQUEST = Item() # approveChatJoinRequest
DECLINE_CHAT_JOIN_REQUEST = Item() # declineChatJoinRequest
SET_CHAT_PHOTO = Item() # setChatPhoto
DELETE_CHAT_PHOTO = Item() # deleteChatPhoto
SET_CHAT_TITLE = Item() # setChatTitle
SET_CHAT_DESCRIPTION = Item() # setChatDescription
PIN_CHAT_MESSAGE = Item() # pinChatMessage
UNPIN_CHAT_MESSAGE = Item() # unpinChatMessage
UNPIN_ALL_CHAT_MESSAGES = Item() # unpinAllChatMessages
LEAVE_CHAT = Item() # leaveChat
GET_CHAT = Item() # getChat
GET_CHAT_ADMINISTRATORS = Item() # getChatAdministrators
GET_CHAT_MEMBER_COUNT = Item() # getChatMemberCount
GET_CHAT_MEMBERS_COUNT = Item() # getChatMembersCount (renamed to getChatMemberCount)
GET_CHAT_MEMBER = Item() # getChatMember
SET_CHAT_STICKER_SET = Item() # setChatStickerSet
DELETE_CHAT_STICKER_SET = Item() # deleteChatStickerSet
GET_FORUM_TOPIC_ICON_STICKERS = Item() # getForumTopicIconStickers
CREATE_FORUM_TOPIC = Item() # createForumTopic
EDIT_FORUM_TOPIC = Item() # editForumTopic
CLOSE_FORUM_TOPIC = Item() # closeForumTopic
REOPEN_FORUM_TOPIC = Item() # reopenForumTopic
DELETE_FORUM_TOPIC = Item() # deleteForumTopic
UNPIN_ALL_FORUM_TOPIC_MESSAGES = Item() # unpinAllForumTopicMessages
EDIT_GENERAL_FORUM_TOPIC = Item() # editGeneralForumTopic
CLOSE_GENERAL_FORUM_TOPIC = Item() # closeGeneralForumTopic
REOPEN_GENERAL_FORUM_TOPIC = Item() # reopenGeneralForumTopic
HIDE_GENERAL_FORUM_TOPIC = Item() # hideGeneralForumTopic
UNHIDE_GENERAL_FORUM_TOPIC = Item() # unhideGeneralForumTopic
ANSWER_CALLBACK_QUERY = Item() # answerCallbackQuery
SET_MY_COMMANDS = Item() # setMyCommands
DELETE_MY_COMMANDS = Item() # deleteMyCommands
GET_MY_COMMANDS = Item() # getMyCommands
# Updating messages
EDIT_MESSAGE_TEXT = Item() # editMessageText
EDIT_MESSAGE_CAPTION = Item() # editMessageCaption
EDIT_MESSAGE_MEDIA = Item() # editMessageMedia
EDIT_MESSAGE_REPLY_MARKUP = Item() # editMessageReplyMarkup
STOP_POLL = Item() # stopPoll
DELETE_MESSAGE = Item() # deleteMessage
# Stickers
SEND_STICKER = Item() # sendSticker
GET_STICKER_SET = Item() # getStickerSet
UPLOAD_STICKER_FILE = Item() # uploadStickerFile
GET_CUSTOM_EMOJI_STICKERS = Item() # getCustomEmojiStickers
CREATE_NEW_STICKER_SET = Item() # createNewStickerSet
ADD_STICKER_TO_SET = Item() # addStickerToSet
SET_STICKER_POSITION_IN_SET = Item() # setStickerPositionInSet
DELETE_STICKER_FROM_SET = Item() # deleteStickerFromSet
SET_STICKER_SET_THUMB = Item() # setStickerSetThumb
# Inline mode
ANSWER_INLINE_QUERY = Item() # answerInlineQuery
ANSWER_WEB_APP_QUERY = Item() # answerWebAppQuery
SET_CHAT_MENU_BUTTON = Item() # setChatMenuButton
GET_CHAT_MENU_BUTTON = Item() # getChatMenuButton
SET_MY_DEFAULT_ADMINISTRATOR_RIGHTS = Item() # setMyDefaultAdministratorRights
GET_MY_DEFAULT_ADMINISTRATOR_RIGHTS = Item() # getMyDefaultAdministratorRights
# Payments
SEND_INVOICE = Item() # sendInvoice
CREATE_INVOICE_LINK = Item() # createInvoiceLink
ANSWER_SHIPPING_QUERY = Item() # answerShippingQuery
ANSWER_PRE_CHECKOUT_QUERY = Item() # answerPreCheckoutQuery
# Telegram Passport
SET_PASSPORT_DATA_ERRORS = Item() # setPassportDataErrors
# Games
SEND_GAME = Item() # sendGame
SET_GAME_SCORE = Item() # setGameScore
GET_GAME_HIGH_SCORES = Item() # getGameHighScores