From 36163690a5d4fd2e3ada4e5330d5879db5fe0002 Mon Sep 17 00:00:00 2001 From: Alex Root Junior Date: Sun, 8 Nov 2020 00:37:59 +0200 Subject: [PATCH] AIOG-T-79: Easy way to use custom API server --- aiogram/bot/api.py | 74 +++++++++++++++++++++++++++------------------ aiogram/bot/base.py | 11 +++++-- 2 files changed, 52 insertions(+), 33 deletions(-) diff --git a/aiogram/bot/api.py b/aiogram/bot/api.py index c83b7ede..03c7e209 100644 --- a/aiogram/bot/api.py +++ b/aiogram/bot/api.py @@ -1,20 +1,57 @@ import logging import os +from dataclasses import dataclass from http import HTTPStatus import aiohttp from .. import types -from ..utils import exceptions -from ..utils import json +from ..utils import exceptions, json from ..utils.helper import Helper, HelperMode, Item # Main aiogram logger log = logging.getLogger('aiogram') -# API Url's -API_URL = "https://api.telegram.org/bot{token}/{method}" -FILE_URL = "https://api.telegram.org/file/bot{token}/{path}" + +@dataclass(frozen=True) +class TelegramAPIServer: + """ + Base config for API Endpoints + """ + + base: str + file: str + + 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) + + 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}}/{{method}}", + ) + + +TELEGRAM_PRODUCTION = TelegramAPIServer.from_base("https://api.telegram.org") def check_token(token: str) -> bool: @@ -92,11 +129,10 @@ def check_result(method_name: str, content_type: str, status_code: int, body: st raise exceptions.TelegramAPIError(f"{description} [{status_code}]") -async def make_request(session, token, method, data=None, files=None, **kwargs): - # log.debug(f"Make request: '{method}' with data: {data} and files {files}") +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 = Methods.api_url(token=token, method=method) + url = server.api_url(token=token, method=method) req = compose_data(data, files) try: @@ -246,25 +282,3 @@ class Methods(Helper): SEND_GAME = Item() # sendGame SET_GAME_SCORE = Item() # setGameScore GET_GAME_HIGH_SCORES = Item() # getGameHighScores - - @staticmethod - def api_url(token, method): - """ - Generate API URL with included token and method name - - :param token: - :param method: - :return: - """ - return API_URL.format(token=token, method=method) - - @staticmethod - def file_url(token, path): - """ - Generate File URL with included token and file path - - :param token: - :param path: - :return: - """ - return FILE_URL.format(token=token, path=path) diff --git a/aiogram/bot/base.py b/aiogram/bot/base.py index d5b036e2..07e44c1c 100644 --- a/aiogram/bot/base.py +++ b/aiogram/bot/base.py @@ -12,6 +12,7 @@ import certifi from aiohttp.helpers import sentinel from . import api +from .api import TelegramAPIServer, TELEGRAM_PRODUCTION from ..types import ParseMode, base from ..utils import json from ..utils.auth_widget import check_integrity @@ -34,7 +35,8 @@ class BaseBot: proxy_auth: Optional[aiohttp.BasicAuth] = None, validate_token: Optional[base.Boolean] = True, parse_mode: typing.Optional[base.String] = None, - timeout: typing.Optional[typing.Union[base.Integer, base.Float, aiohttp.ClientTimeout]] = None + timeout: typing.Optional[typing.Union[base.Integer, base.Float, aiohttp.ClientTimeout]] = None, + server: TelegramAPIServer = TELEGRAM_PRODUCTION ): """ Instructions how to get Bot token is found here: https://core.telegram.org/bots#3-how-do-i-create-a-bot @@ -55,6 +57,8 @@ class BaseBot: :type parse_mode: :obj:`str` :param timeout: Request timeout :type timeout: :obj:`typing.Optional[typing.Union[base.Integer, base.Float, aiohttp.ClientTimeout]]` + :param server: Telegram Bot API Server endpoint. + :type server: :obj:`TelegramAPIServer` :raise: when token is invalid throw an :obj:`aiogram.utils.exceptions.ValidationError` """ self._main_loop = loop @@ -65,6 +69,7 @@ class BaseBot: self._token = None self.__token = token self.id = int(token.split(sep=':')[0]) + self.server = server self.proxy = proxy self.proxy_auth = proxy_auth @@ -200,7 +205,7 @@ class BaseBot: :rtype: Union[List, Dict] :raise: :obj:`aiogram.exceptions.TelegramApiError` """ - return await api.make_request(self.session, self.__token, method, data, files, + return await api.make_request(self.session, self.server, self.__token, method, data, files, proxy=self.proxy, proxy_auth=self.proxy_auth, timeout=self.timeout, **kwargs) async def download_file(self, file_path: base.String, @@ -240,7 +245,7 @@ class BaseBot: return dest def get_file_url(self, file_path): - return api.Methods.file_url(token=self.__token, path=file_path) + return self.server.file_url(token=self.__token, path=file_path) async def send_file(self, file_type, method, file, payload) -> Union[Dict, base.Boolean]: """