Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_retry_timeout(self):
query = self.params
responses.add(
responses.POST,
'https://httpbin.org/post',
json=query,
status=429,
content_type='application/json'
)
client = RouterMock(base_url="https://httpbin.org", retry_over_query_limit=True, retry_timeout=3)
with self.assertRaises(routingpy.exceptions.OverQueryLimit):
client.directions(url='/post', post_params=query)
def test_router_by_name(self):
for s in routingpy.routers._SERVICE_TO_ROUTER.keys():
router = routingpy.routers.get_router_by_name(s)
with self.assertRaises(routingpy.exceptions.RouterNotFound):
router = routingpy.routers.get_router_by_name('orsm')
def _get_body(response):
status_code = response.status_code
try:
body = response.json()
except json.decoder.JSONDecodeError:
raise exceptions.JSONParseError("Can't decode JSON response:{}".format(response.text))
if status_code == 429:
raise exceptions.OverQueryLimit(status_code, body)
if 400 <= status_code < 500:
raise exceptions.RouterApiError(status_code, body)
if 500 <= status_code:
raise exceptions.RouterServerError(status_code, body)
if status_code != 200:
raise exceptions.RouterError(status_code, body)
return body
def _get_body(response):
status_code = response.status_code
try:
body = response.json()
except json.decoder.JSONDecodeError:
raise exceptions.JSONParseError("Can't decode JSON response:{}".format(response.text))
if status_code == 429:
raise exceptions.OverQueryLimit(status_code, body)
if 400 <= status_code < 500:
raise exceptions.RouterApiError(status_code, body)
if 500 <= status_code:
raise exceptions.RouterServerError(status_code, body)
if status_code != 200:
raise exceptions.RouterError(status_code, body)
return body
:raises routingpy.exceptions.RouterError: when anything else happened while requesting.
:raises routingpy.exceptions.JSONParseError: when the JSON response can't be parsed.
:raises routingpy.exceptions.Timeout: when the request timed out.
:raises routingpy.exceptions.TransportError: when something went wrong while trying to
execute a request.
:returns: raw JSON response.
:rtype: dict
"""
if not first_request_time:
first_request_time = datetime.now()
elapsed = datetime.now() - first_request_time
if elapsed > self.retry_timeout:
raise exceptions.Timeout()
if retry_counter > 0:
# 0.5 * (1.5 ^ i) is an increased sleep time of 1.5x per iteration,
# starting at 0.5s when retry_counter=1. The first retry will occur
# at 1, so subtract that first.
delay_seconds = 1.5**(retry_counter - 1)
# Jitter this value by 50% and pause.
time.sleep(delay_seconds * (random.random() + 0.5))
authed_url = self._generate_auth_url(url, get_params)
# Default to the client-level self.requests_kwargs, with method-level
# requests_kwargs arg overriding.
requests_kwargs = requests_kwargs or {}
final_requests_kwargs = dict(self.requests_kwargs, **requests_kwargs)
# Only print URL and parameters for dry_run
if dry_run:
print(
"url:\n{}\nParameters:\n{}".format(
self.base_url + authed_url, json.dumps(final_requests_kwargs, indent=2)
)
)
return
try:
response = requests_method(self.base_url + authed_url, **final_requests_kwargs)
self._req = response.request
except requests.exceptions.Timeout:
raise exceptions.Timeout()
tried = retry_counter + 1
if response.status_code in _RETRIABLE_STATUSES:
# Retry request.
warnings.warn(
'Server down.\nRetrying for the {}{} time.'.format(tried, get_ordinal(tried)),
UserWarning
)
return self._request(
url, get_params, post_params, first_request_time, retry_counter + 1, requests_kwargs
)
try:
result = self._get_body(response)
# Retry request.
warnings.warn(
'Server down.\nRetrying for the {}{} time.'.format(tried, get_ordinal(tried)),
UserWarning
)
return self._request(
url, get_params, post_params, first_request_time, retry_counter + 1, requests_kwargs
)
try:
result = self._get_body(response)
return result
except exceptions.RouterApiError:
if self.skip_api_error:
warnings.warn(
"Router {} returned an API error with "
"the following message:\n{}".format(self.__class__.__name__, response.text)
)
return
raise
except exceptions.RetriableRequest as e:
if isinstance(e, exceptions.OverQueryLimit) and not self.retry_over_query_limit:
raise
warnings.warn(
'Rate limit exceeded.\nRetrying for the {}{} time.'.format(tried, get_ordinal(tried)),
UserWarning