Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if 'HTTP_PROXY' in os.environ and scheme == 'http':
kwargs['proxy'] = os.environ['HTTP_PROXY']
elif 'HTTPS_PROXY' in os.environ and scheme == 'https':
kwargs['proxy'] = os.environ['HTTPS_PROXY']
async with session.get(url, timeout=timeout, **kwargs) as resp:
if resp.status != 200:
raise FailedDownload(filepath_partial, url, resp)
else:
filepath, skip = get_filepath(filepath_partial(resp, url), overwrite)
if skip:
return str(filepath)
if callable(file_pb):
file_pb = file_pb(position=token.n, unit='B', unit_scale=True,
desc=filepath.name, leave=False,
total=get_http_size(resp))
else:
file_pb = None
# This queue will contain the downloaded chunks and their offsets
# as tuples: (offset, chunk)
downloaded_chunk_queue = asyncio.Queue()
download_workers = []
writer = self.loop.create_task(
self._write_worker(downloaded_chunk_queue, file_pb, filepath))
if max_splits and resp.headers.get('Accept-Ranges', None) == "bytes":
content_length = int(resp.headers['Content-length'])
split_length = content_length // max_splits
ranges = [
[start, start + split_length]