in modular_cli/service/adapter_client.py [0:0]
def __make_request(self, resource: str, method: str, payload: dict = None,
params_to_log: dict = None) -> requests.Response:
assert method in self.__method_to_function # todo allow all methods
method_func = self.__method_to_function[method]
parameters = dict(url=f'{self.__api_link}{resource}')
if method == HTTP_GET:
parameters.update(params=payload)
else:
parameters.update(json=payload)
SYSTEM_LOG.debug(
f'API request info: Resource: {resource}; '
f'Parameters: {params_to_log if params_to_log else {}}; '
f'Method: {method}.')
# todo fix the kludges with paths
if self.__token and resource not in ('/login', '/refresh'):
parameters.update(
headers={'authorization': f'Bearer {self.__token}'}
)
elif resource != '/refresh':
parameters.update(auth=(self.__username, self.__secret))
if parameters.get('headers'):
parameters['headers'].update({'Cli-Version': __version__})
else:
parameters['headers'] = {'Cli-Version': __version__}
try:
response = method_func(**parameters)
except requests.exceptions.ConnectTimeout:
message = 'Failed to establish connection with the server due ' \
'to exceeded timeout. Probably a security group ' \
'denied the request'
SYSTEM_LOG.exception(message)
raise ModularCliTimeoutException(message)
except requests.exceptions.ConnectionError:
raise ModularCliConfigurationException(
'Provided configuration api_link is invalid or outdated. '
'Please contact the tool support team.')
SYSTEM_LOG.debug(f'API response info: {response}')
return response