Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""
Download a file to a temporary directory, returning it.
The options provided will control the behaviour of the download algorithm.
* 'tries' - The maximum number of tries to download the file before
giving up and raising an exception.
* 'timeout' - Timeout in seconds before considering the connection to
have failed.
* 'verifier' - A function which is called with a filelike object. It
should return True if the file is okay and appears to be fully
downloaded.
"""
logger = logging.getLogger('download')
with closing(tempfile.NamedTemporaryFile()) as tmp:
# current file position = number of bytes read
filepos = 0
# file size when downloaded, if known
filesize = None
# number of attempts so far
tries = 0
# last try which resulted in some forward progress (i.e: filepos got
# bigger)
last_successful_try = 0
# maximum number of attempts to make
max_tries = options.get('tries', 1)
def build(files, srs):
with closing(tempfile.NamedTemporaryFile(suffix='.vrt')) as vrt:
# ensure files are actually present before trying to make a VRT from
# them.
for f in files:
assert os.path.exists(f), "Trying to build a VRT including file " \
"%r, but it does not seem to exist." % f
args = ["gdalbuildvrt", "-q", "-a_srs", srs, vrt.name ] + files
status = subprocess.call(args)
if status != 0:
raise RuntimeError("Call to gdalbuildvrt failed: status=%r" % status)
ds = gdal.Open(vrt.name)
yield ds
del ds
def download_file(url, filename, session=None, headers=None, **kwargs):
"""Download a file specified.
:param url: Source URL
:param filename: Target file on filesystem
:param session: request session to use
:param headers: override existing headers in request session
:return: True on success, False on failure
"""
try:
hooks, cookies, verify, proxies = request_defaults(kwargs)
with closing(session.get(url, allow_redirects=True, stream=True,
verify=verify, headers=headers, cookies=cookies,
hooks=hooks, proxies=proxies)) as resp:
if not resp.ok:
logger.debug(u"Requested download url %s returned status code is %s: %s"
% (url, resp.status_code, http_code_description(resp.status_code)))
return False
try:
with io.open(filename, 'wb') as fp:
for chunk in resp.iter_content(chunk_size=1024):
if chunk:
fp.write(chunk)
fp.flush()
chmodAsParent(filename)