feat(api): add support for custom API urls following pattern

This commit is contained in:
mpa 2020-11-05 14:35:05 +04:00
parent ae67814abb
commit 888b57b19c
No known key found for this signature in database
GPG key ID: BCCFBFCCC9B754A8
2 changed files with 44 additions and 18 deletions

View file

@ -92,11 +92,19 @@ 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):
async def make_request(
session: aiohttp.ClientSession,
token: str,
method: str,
data=None,
files=None,
url_pattern: str = API_URL,
**kwargs,
):
# log.debug(f"Make request: '{method}' with data: {data} and files {files}")
log.debug('Make request: "%s" with data: "%r" and files "%r"', method, data, files)
url = Methods.api_url(token=token, method=method)
url = Methods.api_url(token=token, method=method, pattern=url_pattern)
req = compose_data(data, files)
try:
@ -244,23 +252,25 @@ class Methods(Helper):
GET_GAME_HIGH_SCORES = Item() # getGameHighScores
@staticmethod
def api_url(token, method):
def api_url(token, method, pattern: str = API_URL) -> str:
"""
Generate API URL with included token and method name
:param token:
:param method:
:param pattern: string with token,method format args
:return:
"""
return API_URL.format(token=token, method=method)
return pattern.format(token=token, method=method)
@staticmethod
def file_url(token, path):
def file_url(token, path, pattern: str = FILE_URL) -> str:
"""
Generate File URL with included token and file path
:param token:
:param path:
:param pattern: string with token,path format args
:return:
"""
return FILE_URL.format(token=token, path=path)
return pattern.format(token=token, path=path)

View file

@ -25,15 +25,16 @@ class BaseBot:
_ctx_token = ContextVar('BotDifferentToken')
def __init__(
self,
token: base.String,
loop: Optional[Union[asyncio.BaseEventLoop, asyncio.AbstractEventLoop]] = None,
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: typing.Optional[base.String] = None,
timeout: typing.Optional[typing.Union[base.Integer, base.Float, aiohttp.ClientTimeout]] = None
self,
token: base.String,
loop: Optional[Union[asyncio.BaseEventLoop, asyncio.AbstractEventLoop]] = None,
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: typing.Optional[base.String] = None,
timeout: typing.Optional[typing.Union[base.Integer, base.Float, aiohttp.ClientTimeout]] = None,
api_url_configs: typing.Tuple[str, str] = (api.API_URL, api.FILE_URL),
):
"""
Instructions how to get Bot token is found here: https://core.telegram.org/bots#3-how-do-i-create-a-bot
@ -54,6 +55,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 api_url_configs: tuple of METHOD,FILE urls
:type api_url_configs: :obj:`typing.Tuple[str, str]`
:raise: when token is invalid throw an :obj:`aiogram.utils.exceptions.ValidationError`
"""
self._main_loop = loop
@ -71,6 +74,7 @@ class BaseBot:
# aiohttp main session
ssl_context = ssl.create_default_context(cafile=certifi.where())
self._api_url, self._file_url = api_url_configs
self._session: Optional[aiohttp.ClientSession] = None
self._connector_class: Type[aiohttp.TCPConnector] = aiohttp.TCPConnector
self._connector_init = dict(limit=connections_limit, ssl=ssl_context)
@ -197,8 +201,18 @@ class BaseBot:
:rtype: Union[List, Dict]
:raise: :obj:`aiogram.exceptions.TelegramApiError`
"""
return await api.make_request(self.session, self.__token, method, data, files,
proxy=self.proxy, proxy_auth=self.proxy_auth, timeout=self.timeout, **kwargs)
return await api.make_request(
self.session,
self.__token,
method,
data,
files,
proxy=self.proxy,
proxy_auth=self.proxy_auth,
timeout=self.timeout,
url_pattern=self._api_url,
**kwargs,
)
async def download_file(self, file_path: base.String,
destination: Optional[base.InputFile] = None,
@ -237,7 +251,9 @@ class BaseBot:
return dest
def get_file_url(self, file_path):
return api.Methods.file_url(token=self.__token, path=file_path)
return api.Methods.file_url(
token=self.__token, path=file_path, pattern=self._file_url,
)
async def send_file(self, file_type, method, file, payload) -> Union[Dict, base.Boolean]:
"""