Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def download_zip(self, destination, progress_callback=None, parallel_downloads=16, parallel_chunks_size=10):
"""Download this task's assets archive to a directory.
Args:
destination (str): directory where to download assets archive. If the directory does not exist, it will be created.
progress_callback (function): an optional callback with one parameter, the download progress percentage.
parallel_downloads (int): maximum number of parallel downloads if the node supports http range.
parallel_chunks_size (int): size in MB of chunks for parallel downloads
Returns:
str: path to archive file (.zip)
"""
info = self.info()
if info.status != TaskStatus.COMPLETED:
raise NodeResponseError("Cannot download task, task status is " + str(info.status))
if not os.path.exists(destination):
os.makedirs(destination, exist_ok=True)
try:
download_stream = self.get('/task/{}/download/all.zip'.format(self.uuid), stream=True)
headers = download_stream.headers
zip_path = os.path.join(destination, "{}_{}_all.zip".format(self.uuid, int(time.time())))
# Keep track of download progress (if possible)
content_length = download_stream.headers.get('content-length')
total_length = int(content_length) if content_length is not None else None
downloaded = 0
chunk_size = int(parallel_chunks_size * 1024 * 1024)
from pyodm.types import TaskStatus
QUEUED = TaskStatus.QUEUED.value
RUNNING = TaskStatus.RUNNING.value
FAILED = TaskStatus.FAILED.value
COMPLETED = TaskStatus.COMPLETED.value
CANCELED = TaskStatus.CANCELED.value