AIOG-T-79: Easy way to use custom API server

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)

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]:
         """