in scripts/health_check.py [0:0]
def _fetch(self, path: str):
url = f'{self.environment.value}/{path}'
last_exception = None
for attempt in range(1, self.retries + 1):
self.log(
f'Attempt {attempt}/{self.retries}: '
f'Requesting {url} for {self.environment.name}'
)
try:
req = urllib.request.Request(url)
with urllib.request.urlopen(req, timeout=self.timeout) as response:
return self._response(response)
except (
urllib.error.HTTPError,
urllib.error.URLError,
http.client.RemoteDisconnected,
TimeoutError,
) as e:
last_exception = e
should_retry = False
log_reason = ''
if isinstance(e, urllib.error.HTTPError):
log_reason = f'status {e.code}'
try:
self.log(
f'Request failed with {log_reason}, '
'attempting to parse error response body.'
)
return self._response(e)
except (
json.decoder.JSONDecodeError,
UnicodeDecodeError,
) as parse_error:
self.log(f'Failed to parse error response body: {parse_error}')
if e.code in self.status_forcelist and attempt < self.retries:
should_retry = True
else:
log_reason = str(e)
if attempt < self.retries:
should_retry = True
if should_retry:
wait_time = self.backoff_factor * (2**attempt)
self.log(
f'Request failed with {log_reason}. '
f'Retrying in {wait_time:.2f} seconds...'
)
time.sleep(wait_time)
continue
else:
self.log(
f'Request failed with {log_reason}. '
'No more retries or not retryable.'
)
raise e
except Exception as e:
last_exception = e
self.log(f'An unexpected error occurred: {e}. No more retries.')
raise e
raise last_exception or RuntimeError(
f'Failed to fetch {url} after {self.retries + 1} attempts'
)