Multiple-threaded upload/downloads (#109)

* add `threads` argument to backend interface

* Update S3Client to use TransferConfig in up/downloads

TransferConfig allows multi-threaded uploads, and also hides multipart uploads, so we don't implement that on our layer anymore.

* add threads to options file

* remove unused import in options

* add optional `threads` argument to I/O functions in cottoncandy interfaces to allow setting thread count dynamically at runtime

* update gdriveclient to have `threads` argument for interface consistency

* upload localclient to have `threads` for interface consistency

* threads in docstrings

* remove print debug statement

* fix missing comma

* tabs -> spaces

Co-authored-by: Copilot <[email protected]>

* Add ACL to upload_file

Co-authored-by: Copilot <[email protected]>

* Update cottoncandy/s3client.py

Co-authored-by: Copilot <[email protected]>

* copilot docstring suggestion

Co-authored-by: Copilot <[email protected]>

---------

Co-authored-by: Copilot <[email protected]>

Author: candytaco

cottoncandy/backend.py

@@ -30,11 +30,12 @@ def check_file_exists(self, file_name, bucket_name):
         pass
 
     @abstractmethod
-    def upload_stream(self, stream, cloud_name, metadata, permissions):
+    def upload_stream(self, stream, cloud_name, metadata, permissions, threads):
         """Uploads a stream object with a .read() function
 
         Parameters
         ----------
+        threads
         stream : stream
             streaming object
         cloud_name : str
@@ -43,6 +44,8 @@ def upload_stream(self, stream, cloud_name, metadata, permissions):
             custom metadata for this file
         permissions : str?
             permissions for this file
+        threads : int
+            number of threads to use
 
         Returns
         -------
@@ -51,57 +54,39 @@ def upload_stream(self, stream, cloud_name, metadata, permissions):
         pass
 
     @abstractmethod
-    def upload_file(self, file_name, cloud_name, permissions):
+    def upload_file(self, file_name, cloud_name, permissions, threads):
         """Uploads a file from disk
 
         Parameters
         ----------
+        threads
         file_name : str
             name of file to upload
         cloud_name : str
             name to use on the cloud
         permissions : str?
             permissions for this file
+        threads : int
+            number of threads to use
 
         Returns
         -------
         bool, upload success
         """
         pass
 
-    @abstractmethod
-    def upload_multipart(self, stream, cloud_name, metadata, permissions, buffersize, verbose):
-        """Multi-part upload for large stream objects
-
-        Parameters
-        ----------
-        stream : stream
-            streaming object
-        cloud_name : str
-            name to use on cloud
-        metadata : dict
-            custom metadata
-        permissions : str?
-            permissions for this file
-        buffersize : int
-            s3 uploading buffersize
-        verbose : bool
-            s3 verbosity
-
-        Returns
-        -------
-        bool, upload success
-        """
-        pass
 
     @abstractmethod
-    def download_stream(self, cloud_name):
+    def download_stream(self, cloud_name, threads):
         """Downloads a object to an in-memory stream
 
         Parameters
         ----------
+        threads
         cloud_name : str
             name of object to download
+        threads : int
+            number of threads to use
 
         Returns
         -------
@@ -110,15 +95,18 @@ def download_stream(self, cloud_name):
         pass
 
     @abstractmethod
-    def download_to_file(self, cloud_name, file_name):
+    def download_to_file(self, cloud_name, file_name, threads):
         """Downloads an object directly to disk
 
         Parameters
         ----------
+        threads
         cloud_name : str
             name of object to download
         file_name : str
             name on disk to use
+        threads : int
+            number of threads to use
 
         Returns
         -------

cottoncandy/defaults.cfg

@@ -12,6 +12,7 @@ mandatory_bucket_prefix =
 force_bucket_creation = False
 path_separator = /
 signature_version =
+threads = 8
 
 [upload_settings]
 # in MB, except max_mpu_size_TB, and max_mpu_parts

cottoncandy/gdriveclient.py

@@ -411,7 +411,7 @@ def delete(self, file_name, recursive=False, delete=False):
 
     #### File IO functions
 
-    def upload_file(self, file_name, cloud_name=None, permissions=None):
+    def upload_file(self, file_name, cloud_name=None, permissions=None, threads = 1):
         """Uploads from file on disk
 
         Parameters
@@ -468,25 +468,22 @@ def upload_file(self, file_name, cloud_name=None, permissions=None):
 
         return True
 
-    def upload_stream(self, stream, name, properties=None, permissions=None):
+    def upload_stream(self, stream, cloud_name, properties, permissions, threads = 1):
         """Upload a stream with a .read() method
 
         Parameters
         ----------
+        threads
         stream : stream
             stream to upload
-        name : str
-            name to use for cloud file
-        properties : dict
-            custom metadata
 
         Returns
         -------
         : bool
             success of operation
         """
         # cloudName formatting
-        name = re.sub('^./', '', name)
+        name = re.sub('^./', '', cloud_name)
 
         # not current dir
         current_directory = None
@@ -524,11 +521,8 @@ def upload_stream(self, stream, name, properties=None, permissions=None):
 
         return True
 
-    def upload_multipart(self, stream, cloud_name, properties=None, permissions=None, buffersize = -1, verbose = False):
-        return self.upload_stream(stream, cloud_name, properties, permissions)
-
 
-    def download_to_file(self, drive_file, local_file=None):
+    def download_to_file(self, drive_file, local_file=None, threads = 1):
         """Download a file to disk
 
         Parameters
@@ -557,7 +551,7 @@ def download_to_file(self, drive_file, local_file=None):
 
         return True
 
-    def download_stream(self, drive_file):
+    def download_stream(self, drive_file, threads = 1):
         """Downloads a file to memory
 
         Parameters