Merge pull request #295 from Duke-GCB/simplify-downloading

simplify downloading

Author: johnbradley

ddsc/config.py

@@ -24,6 +24,7 @@
 FILE_EXCLUDE_REGEX_DEFAULT = '^\.DS_Store$|^\.ddsclient$|^\.\_'
 MAX_DEFAULT_WORKERS = 8
 GET_PAGE_SIZE_DEFAULT = 100  # fetch 100 items per page
+DEFAULT_FILE_DOWNLOAD_RETRIES = 5
 
 
 def get_user_config_filename():
@@ -74,6 +75,7 @@ class Config(object):
     FILE_EXCLUDE_REGEX = 'file_exclude_regex'          # allows customization of which filenames will be uploaded
     GET_PAGE_SIZE = 'get_page_size'                    # page size used for GET pagination requests
     STORAGE_PROVIDER_ID = 'storage_provider_id'        # setting to override the default storage provider
+    FILE_DOWNLOAD_RETRIES = 'file_download_retries'    # number of times to retry a failed file download
 
     def __init__(self):
         self.values = {}
@@ -160,8 +162,7 @@ def download_workers(self):
         Return the number of parallel works to use when downloading a file.
         :return: int number of workers. Specify None or 1 to disable parallel downloading
         """
-        # Profiling download on different servers showed half the number of CPUs to be optimum for speed.
-        default_workers = int(math.ceil(default_num_workers() / 2))
+        default_workers = int(math.ceil(default_num_workers()))
         return self.values.get(Config.DOWNLOAD_WORKERS, default_workers)
 
     @property
@@ -224,3 +225,11 @@ def storage_provider_id(self):
         :return: str: uuid of storage provider
         """
         return self.values.get(Config.STORAGE_PROVIDER_ID, None)
+
+    @property
+    def file_download_retries(self):
+        """
+        Returns number of times to retry failed external file downloads
+        :return: int: number of retries allowed before failure
+        """
+        return self.values.get(Config.FILE_DOWNLOAD_RETRIES, DEFAULT_FILE_DOWNLOAD_RETRIES)

ddsc/core/d4s2.py

@@ -8,11 +8,12 @@
 import tempfile
 import requests
 from ddsc.core.upload import ProjectUpload
-from ddsc.core.download import ProjectDownload
+from ddsc.core.download import ProjectFileDownloader
 from ddsc.core.ddsapi import DataServiceAuth
 from ddsc.core.util import KindType
 from ddsc.versioncheck import get_internal_version_str
 from ddsc.core.remotestore import ProjectNameOrId, RemotePath
+from ddsc.sdk.client import Client
 
 UNAUTHORIZED_MESSAGE = """
 ERROR: Your account does not have authorization for D4S2 (the deliver/share service).
@@ -242,6 +243,7 @@ def __init__(self, config, remote_store, print_func):
         :param print_func: func used to print output somewhere
         """
         self.config = config
+        self.client = Client(config)
         auth = DataServiceAuth(self.config)
         api_token = auth.get_auth()
         self.api = D4S2Api(config.d4s2_url, api_token)
@@ -345,24 +347,32 @@ def _copy_project(self, project, new_project_name, path_filter):
         if remote_project:
             raise ValueError("A project with name '{}' already exists.".format(new_project_name))
         activity = CopyActivity(self.remote_store.data_service, project, new_project_name)
-        self._download_project(activity, project, temp_directory, path_filter)
+        self._download_project(activity, project.id, temp_directory, path_filter)
         self._upload_project(activity, new_project_name, temp_directory)
         activity.finished()
         shutil.rmtree(temp_directory)
         return self.remote_store.fetch_remote_project(new_project_name_or_id, must_exist=True)
 
-    def _download_project(self, activity, project, temp_directory, path_filter):
+    def _download_project(self, activity, project_id, temp_directory, path_filter):
         """
         Download the project with project_name to temp_directory.
         :param activity: CopyActivity: info about the copy activity are downloading for
-        :param project: remotestore.RemoteProject project to download
+        :param project_id: uuid of the project to download
         :param temp_directory: str path to directory we can download into
         :param path_filter: PathFilter: filters what files are shared
         """
+        project = self.client.get_project_by_id(project_id)
         self.print_func("Downloading a copy of '{}'.".format(project.name))
-        project_download = ProjectDownload(self.remote_store, project, temp_directory, path_filter,
-                                           file_download_pre_processor=DownloadedFileRelations(activity))
-        project_download.run()
+
+        downloader = ProjectFileDownloader(self.config, temp_directory, project, path_filter)
+        downloader.run()
+
+        downloaded_file_relations = DownloadedFileRelations(activity)
+        for remote_path, dds_file in project.get_path_to_files().items():
+            if path_filter.include_path(remote_path):
+                downloaded_file_relations.add(self.remote_store.data_service,
+                                              remote_path,
+                                              dds_file.current_version['id'])
 
     def _upload_project(self, activity, project_name, temp_directory):
         """
@@ -435,15 +445,10 @@ def __init__(self, activity):
         """
         self.activity = activity
 
-    def run(self, data_service, project_file):
+    def add(self, data_service, remote_path, file_version_id):
         """
         Attach a remote file to activity with used relationship.
-        :param data_service: DataServiceApi: service used to attach relationship
-        :param project_file: ProjectFile: contains details about a file we will attach
         """
-        remote_path = project_file.path
-        file_dict = data_service.get_file(project_file.id).json()
-        file_version_id = file_dict['current_version']['id']
         data_service.create_used_relation(self.activity.id, KindType.file_str, file_version_id)
         self.activity.remote_path_to_file_version_id[remote_path] = file_version_id
 

ddsc/core/ddsapi.py

@@ -16,6 +16,8 @@
 To cancel this operation, press Ctrl+C.
 """
 CONNECTION_RETRY_MESSAGE = "Connection failed. Retrying."
+DDS_TOTAL_PAGES_HEADER = 'x-total-pages'
+DDS_TOTAL_HEADER = 'x-total'
 
 
 def get_user_agent_str():
@@ -339,7 +341,7 @@ def _get_single_item(self, url_suffix, data, content_type=ContentType.json):
         return self._check_err(resp, url_suffix, data, allow_pagination=False)
 
     @retry_connection_exceptions
-    def _get_single_page(self, url_suffix, data, page_num):
+    def _get_single_page(self, url_suffix, data, page_num, page_size=None):
         """
         Send GET request to API at url_suffix with post_data adding page and per_page parameters to
         retrieve a single page. Page size is determined by config.page_size.
@@ -350,7 +352,9 @@ def _get_single_page(self, url_suffix, data, page_num):
         """
         data_with_per_page = dict(data)
         data_with_per_page['page'] = page_num
-        data_with_per_page['per_page'] = self._get_page_size()
+        if not page_size:
+            page_size = self._get_page_size()
+        data_with_per_page['per_page'] = page_size
         (url, data_str, headers) = self._url_parts(url_suffix, data_with_per_page,
                                                    content_type=ContentType.form)
         resp = self.http.get(url, headers=headers, params=data_str)
@@ -558,6 +562,26 @@ def get_project_files(self, project_id):
         url_prefix = "/projects/{}/files".format(project_id)
         return self._get_collection(url_prefix, {})
 
+    def get_project_files_generator(self, project_id, page_size):
+        """
+        Send GET to /projects/{project_id}/files
+        :param project_id: str uuid of the project
+        :param page_size: int page size to fetch
+        :return: generator that returns (DDS file download JSON, header metadata) pairs
+        """
+        url_suffix = "/projects/{}/files".format(project_id)
+        data = {}
+        # process first page separately to read total pages
+        response = self._get_single_page(url_suffix, data, page_size=page_size, page_num=1)
+        total_pages = int(response.headers.get('x-total-pages'))
+        for item in response.json()["results"]:
+            yield item, response.headers
+        # process the rest of the pages
+        for page in range(2, total_pages + 1):
+            response = self._get_single_page(url_suffix, data, page_size=page_size, page_num=page)
+            for item in response.json()["results"]:
+                yield item, response.headers
+
     def get_folder_children(self, folder_id, name_contains):
         """
         Send GET to /folders/{folder_id} filtering by a name.