Makes a request to the Telegram API. :param token: The bot's API token. (Created with @BotFather) :param method_name: Name of the API method to be called. (E.g. 'getUpdates') :param method: HTTP method to be used. Defaults to 'get'. :param params: Optional parameters. Should be
(token, method_name, method='get', params=None, files=None)
| 68 | |
| 69 | |
| 70 | def _make_request(token, method_name, method='get', params=None, files=None): |
| 71 | """ |
| 72 | Makes a request to the Telegram API. |
| 73 | :param token: The bot's API token. (Created with @BotFather) |
| 74 | :param method_name: Name of the API method to be called. (E.g. 'getUpdates') |
| 75 | :param method: HTTP method to be used. Defaults to 'get'. |
| 76 | :param params: Optional parameters. Should be a dictionary with key-value pairs. |
| 77 | :param files: Optional files. |
| 78 | :return: The result parsed to a JSON dictionary. |
| 79 | """ |
| 80 | if not token: |
| 81 | raise Exception('Bot token is not defined') |
| 82 | if API_URL: |
| 83 | # noinspection PyUnresolvedReferences |
| 84 | request_url = API_URL.format(token, method_name) |
| 85 | else: |
| 86 | request_url = "https://api.telegram.org/bot{0}/{1}".format(token, method_name) |
| 87 | |
| 88 | logger.debug("Request: method={0} url={1} params={2} files={3}".format(method, request_url, params, files).replace(token, token.split(':')[0] + ":{TOKEN}")) |
| 89 | read_timeout = READ_TIMEOUT |
| 90 | connect_timeout = CONNECT_TIMEOUT |
| 91 | |
| 92 | if files: |
| 93 | files_copy = dict(files) |
| 94 | # process types.InputFile |
| 95 | for key, value in files_copy.items(): |
| 96 | if isinstance(value, types.InputFile): |
| 97 | files[key] = (value.file_name, value.file) |
| 98 | elif isinstance(value, tuple) and (len(value) == 2) and isinstance(value[1], types.InputFile): |
| 99 | files[key] = (value[0], value[1].file) |
| 100 | |
| 101 | |
| 102 | if files and format_header_param: |
| 103 | fields.format_header_param = _no_encode(format_header_param) |
| 104 | if params: |
| 105 | if 'timeout' in params: |
| 106 | read_timeout = params.pop('timeout') |
| 107 | connect_timeout = read_timeout |
| 108 | if 'long_polling_timeout' in params: |
| 109 | # For getUpdates. It's the only function with timeout parameter on the BOT API side |
| 110 | long_polling_timeout = params.pop('long_polling_timeout') |
| 111 | params['timeout'] = long_polling_timeout |
| 112 | # Long polling hangs for a given time. Read timeout should be greater that long_polling_timeout |
| 113 | read_timeout = max(long_polling_timeout + 5, read_timeout) |
| 114 | |
| 115 | params = params or None # Set params to None if empty |
| 116 | result = None |
| 117 | |
| 118 | if CUSTOM_REQUEST_SENDER: |
| 119 | # noinspection PyCallingNonCallable |
| 120 | result = CUSTOM_REQUEST_SENDER( |
| 121 | method, request_url, params=params, files=files, |
| 122 | timeout=(connect_timeout, read_timeout), proxies=proxy) |
| 123 | elif RETRY_ON_ERROR and RETRY_ENGINE == 1: |
| 124 | got_result = False |
| 125 | current_try = 0 |
| 126 | while not got_result and current_try<MAX_RETRIES-1: |
| 127 | current_try+=1 |
no test coverage detected
searching dependent graphs…