the uploader supports progress.

Author: zhuxiaolong37

alibabacloud_oss_v2/checkpoint.py

@@ -265,7 +265,7 @@ def __init__(
         h.update(absfilepath.encode())
         src_hash = h.hexdigest()
 
-        if len(basedir) == 0:
+        if basedir is None or len(basedir) == 0:
             dirbase = gettempdir()
         else:
             dirbase = os.path.dirname(basedir)

alibabacloud_oss_v2/uploader.py

@@ -439,6 +439,7 @@ def adjust_source(self):
         self._reader_pos = next_offset
         self._part_number = part_number + 1
         self._ccrc = ccrc
+        self._transferred = next_offset
 
 
     def set_reader(self, reader) ->IO[bytes]:
@@ -691,6 +692,10 @@ def _update_upload_result(self, result):
         if self._check_crc and hash_crc64 is not None:
             self._ccrc = Crc64.combine(self._ccrc, int(hash_crc64), size)
 
+        self._transferred += size
+        if self._request.progress_fn is not None:
+            self._request.progress_fn(size, self._transferred, self._total_size)
+
 
     def _assert_crc_same(self, headers: MutableMapping):
         if not self._check_crc:

sample/progress_uploader_file.py

@@ -0,0 +1,58 @@
+import argparse
+import alibabacloud_oss_v2 as oss
+
+parser = argparse.ArgumentParser(description="progress upload file sample")
+parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
+parser.add_argument('--bucket', help='The name of the bucket.', required=True)
+parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
+parser.add_argument('--key', help='The name of the object.', required=True)
+parser.add_argument('--file_path', help='The path of Upload file.', required=True)
+
+
+class UploadProgress:
+    def __init__(self):
+        self.bytes_transferred = 0
+
+    def __call__(self, increment, written, total):
+        self.bytes_transferred += increment
+        rate = int(100 * (float(written) / float(total)))
+        print(f'\r{rate}% ', end='')
+
+
+def main():
+
+    args = parser.parse_args()
+
+    # Loading credentials values from the environment variables
+    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
+
+    # Using the SDK's default configuration
+    cfg = oss.config.load_default()
+    cfg.credentials_provider = credentials_provider
+    cfg.region = args.region
+    if args.endpoint is not None:
+        cfg.endpoint = args.endpoint
+
+    client = oss.Client(cfg)
+
+    up_loader = client.uploader(part_size=100*1024,
+                                    parallel_num=5,
+                                    leave_parts_on_error=True,
+                                    enable_checkpoint=True,
+                                )
+
+    # Create progress tracker
+    progress_tracker = UploadProgress()
+
+    result = up_loader.upload_file(oss.PutObjectRequest(
+        bucket=args.bucket,
+        key=args.key,
+        progress_fn=progress_tracker,
+    ), filepath=args.file_path)
+
+    print(vars(result))
+
+
+
+if __name__ == "__main__":
+    main()

sample/progress_uploader_from.py

@@ -0,0 +1,60 @@
+import argparse
+import alibabacloud_oss_v2 as oss
+
+parser = argparse.ArgumentParser(description="progress upload from sample")
+parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
+parser.add_argument('--bucket', help='The name of the bucket.', required=True)
+parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
+parser.add_argument('--key', help='The name of the object.', required=True)
+parser.add_argument('--file_path', help='The path of Upload file.', required=True)
+
+
+class UploadProgress:
+    def __init__(self):
+        self.bytes_transferred = 0
+
+    def __call__(self, increment, written, total):
+        self.bytes_transferred += increment
+        rate = int(100 * (float(written) / float(total)))
+        print(f'\r{rate}% ')
+
+
+def main():
+
+    args = parser.parse_args()
+
+    # Loading credentials values from the environment variables
+    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
+
+    # Using the SDK's default configuration
+    cfg = oss.config.load_default()
+    cfg.credentials_provider = credentials_provider
+    cfg.region = args.region
+    if args.endpoint is not None:
+        cfg.endpoint = args.endpoint
+
+    client = oss.Client(cfg)
+
+    up_loader = client.uploader()
+
+    # up_loader = client.uploader(part_size=100*1024,
+    #                                 parallel_num=5,
+    #                                 leave_parts_on_error=True,
+    #                                 enable_checkpoint=True,
+    #                                 checkpoint_dir=args.file_path)
+
+    # Create progress tracker
+    progress_tracker = UploadProgress()
+
+    with open(file=args.file_path, mode='rb') as f:
+        result = up_loader.upload_from(oss.PutObjectRequest(
+            bucket=args.bucket,
+            key=args.key,
+            progress_fn=progress_tracker,
+        ), reader=f)
+
+        print(vars(result))
+
+if __name__ == "__main__":
+    main()
+

tests/integration/test_client.py

@@ -4500,3 +4500,140 @@ def _progress_fn(n, written, total):
         self.assertEqual("Multipart", hresult.object_type)
         self.assertEqual(src_crc64, hresult.hash_crc64)
 
+
+class TestUploader(TestIntegration):
+    def test_uploader_progress_with_single_upload(self):
+        length = 100 * 1024 + 123
+        data = random_str(length)
+        object_name = OBJECTNAME_PREFIX + random_str(16)
+
+        global bytes_added, total_bytes_transferred, total_bytes_expected, last_written
+        bytes_added = 0
+        total_bytes_transferred = 0
+        total_bytes_expected = 0
+        last_written = 0
+
+        def _progress_fn(n, written, total):
+            global last_written
+            global bytes_added
+            global total_bytes_transferred
+            global total_bytes_expected
+
+            n = written - last_written
+            bytes_added += n
+            total_bytes_transferred = written
+            last_written = written
+            total_bytes_expected = total
+
+        uploader = self.client.uploader()
+
+        result = uploader.upload_from(oss.PutObjectRequest(
+                bucket=self.bucket_name,
+                key=object_name,
+                progress_fn=_progress_fn
+            ), io.StringIO(data),
+                part_size=500 * 1024,
+                parallel_num=5,
+                leave_parts_on_error=True
+        )
+        self.assertIsNotNone(result)
+        self.assertEqual(200, result.status_code)
+        self.assertEqual(total_bytes_transferred, length)
+        self.assertEqual(total_bytes_expected, length)
+
+        hresult = self.client.head_object(oss.HeadObjectRequest(
+            bucket=self.bucket_name,
+            key=object_name
+        ))
+        self.assertEqual("Normal", hresult.object_type)
+        self.assertEqual(length, hresult.content_length)
+
+    def test_uploader_progress_with_single_multipart(self):
+        length = 500 * 1024 + 123
+        data = random_str(length)
+        object_name = OBJECTNAME_PREFIX + random_str(16)
+
+        global bytes_added, total_bytes_transferred, total_bytes_expected, last_written
+        bytes_added = 0
+        total_bytes_transferred = 0
+        total_bytes_expected = 0
+        last_written = 0
+
+        def _progress_fn(n, written, total):
+            global last_written
+            global bytes_added
+            global total_bytes_transferred
+            global total_bytes_expected
+
+            n = written - last_written
+            bytes_added += n
+            total_bytes_transferred = written
+            last_written = written
+            total_bytes_expected = total
+
+        uploader = self.client.uploader()
+        result = uploader.upload_from(oss.PutObjectRequest(
+                bucket=self.bucket_name,
+                key=object_name,
+                progress_fn=_progress_fn
+            ), io.StringIO(data),
+                part_size=100 * 1024,
+                parallel_num=1,
+                leave_parts_on_error=True,
+        )
+        self.assertIsNotNone(result)
+        self.assertEqual(200, result.status_code)
+        self.assertEqual(total_bytes_transferred, length)
+        self.assertEqual(total_bytes_expected, length)
+
+        hresult = self.client.head_object(oss.HeadObjectRequest(
+            bucket=self.bucket_name,
+            key=object_name
+        ))
+        self.assertEqual("Multipart", hresult.object_type)
+        self.assertEqual(length, hresult.content_length)
+
+    def test_uploader_progress_with_thread_multipart(self):
+        length = 500 * 1024 + 123
+        data = random_str(length)
+        object_name = OBJECTNAME_PREFIX + random_str(16)
+
+        global bytes_added, total_bytes_transferred, total_bytes_expected, last_written
+        bytes_added = 0
+        total_bytes_transferred = 0
+        total_bytes_expected = 0
+        last_written = 0
+
+        def _progress_fn(n, written, total):
+            global last_written
+            global bytes_added
+            global total_bytes_transferred
+            global total_bytes_expected
+
+            n = written - last_written
+            bytes_added += n
+            total_bytes_transferred = written
+            last_written = written
+            total_bytes_expected = total
+
+        uploader = self.client.uploader()
+        result = uploader.upload_from(oss.PutObjectRequest(
+                bucket=self.bucket_name,
+                key=object_name,
+                progress_fn=_progress_fn
+            ), io.StringIO(data),
+                part_size=100 * 1024,
+                parallel_num=3,
+                leave_parts_on_error=True,
+        )
+        self.assertIsNotNone(result)
+        self.assertEqual(200, result.status_code)
+        self.assertEqual(total_bytes_transferred, length)
+        self.assertEqual(total_bytes_expected, length)
+
+        hresult = self.client.head_object(oss.HeadObjectRequest(
+            bucket=self.bucket_name,
+            key=object_name
+        ))
+        self.assertEqual("Multipart", hresult.object_type)
+        self.assertEqual(length, hresult.content_length)