Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@backoff.on_exception(backoff.expo, urllib.error.URLError, max_tries=5)
def urlopen_retry(uri):
"""Retry urlopen on exception."""
kwargs = {}
try:
# trust the system's default CA certificates
# proper way for 2.7.9+ on Linux
if uri.startswith("https://"):
kwargs['context'] = ssl.create_default_context(
ssl.Purpose.CLIENT_AUTH
)
except AttributeError:
pass
return urllib.request.urlopen(uri, **kwargs)
Path where the file will be saved.
"""
# Convert a local path to a URI
url = watchmaker.utils.uri_from_filepath(url)
self.log.debug('Downloading: %s', url)
self.log.debug('Destination: %s', filename)
try:
self.log.debug('Establishing connection to the host, %s', url)
response = watchmaker.utils.urlopen_retry(url)
self.log.debug('Opening the file handle, %s', filename)
with open(filename, 'wb') as outfile:
self.log.debug('Saving file to local filesystem...')
shutil.copyfileobj(response, outfile)
except (ValueError, urllib.error.URLError):
self.log.critical(
'Failed to retrieve the file. url = %s. filename = %s',
url, filename
)
raise
self.log.info(
'Retrieved the file successfully. url=%s. filename=%s',
url, filename
)
def urlopen_retry(uri):
"""Retry urlopen on exception."""
kwargs = {}
try:
# trust the system's default CA certificates
# proper way for 2.7.9+ on Linux
if uri.startswith("https://"):
kwargs['context'] = ssl.create_default_context(
ssl.Purpose.CLIENT_AUTH
)
except AttributeError:
pass
return urllib.request.urlopen(uri, **kwargs)