Add copier progress

Author: zhuxiaolong37

alibabacloud_oss_v2/copier.py

@@ -265,6 +265,9 @@ def __init__(
         self._total_size = 0
         self._transferred = 0
 
+        parallel = options.parallel_num > 1 and self._request.progress_fn is not None
+        self._progress_lock = threading.Lock() if parallel else None
+
         #Source's Info
         self._metadata_prop = metadata_prop
         self._tag_prop = tag_prop
@@ -330,6 +333,8 @@ def copy(self) -> CopyResult:
     def _single_copy(self) -> CopyResult:
         result = self._client.copy_object(self._request)
 
+        self._update_progress(self._total_size)
+
         ret = CopyResult(
             etag=result.etag,
             version_id=result.version_id,
@@ -352,6 +357,8 @@ def _shallow_copy(self) -> (CopyResult):
                 return self._multipart_copy()
             raise
 
+        self._update_progress(self._total_size)
+
         ret = CopyResult(
             etag=result.etag,
             version_id=result.version_id,
@@ -488,7 +495,7 @@ def _iter_part(self):
             self._reader_pos += n
 
             start_part_num += 1
-            yield upload_id, start_part_num, range, timeout
+            yield upload_id, start_part_num, range, timeout, n
 
     def _copy_part(self, part):
         # When an error occurs, ignore other upload requests
@@ -499,6 +506,7 @@ def _copy_part(self, part):
         part_number = part[1]
         range = part[2]
         timeout = part[3]
+        part_size = part[4]
         error: Exception = None
         etag = None
 
@@ -511,11 +519,25 @@ def _copy_part(self, part):
             result = self._client.upload_part_copy(request, readwrite_timeout=timeout)
             etag = result.etag
 
+            self._update_progress(part_size)
         except Exception as err:
             error = err
 
+
         return part_number, etag, error
 
+    def _update_progress(self, increment: int):
+        if self._request.progress_fn is None:
+            return
+
+        if self._progress_lock:
+            with self._progress_lock:
+                self._transferred += increment
+                self._request.progress_fn(increment, self._transferred, self._total_size)
+        else:
+            self._transferred += increment
+            self._request.progress_fn(increment, self._transferred, self._total_size)
+
     def _update_upload_result_lock(self, result) -> None:
         if self._copy_part_lock:
             with self._copy_part_lock:

sample/progress_copier.py

@@ -0,0 +1,55 @@
+import argparse
+import alibabacloud_oss_v2 as oss
+
+parser = argparse.ArgumentParser(description="progress copier 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('--source_key', help='The name of the source address for object.', required=True)
+parser.add_argument('--source_bucket', help='The name of the source address for bucket.', required=True)
+
+
+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)
+
+    global progress_save_n
+    progress_save_n = 0
+    def _progress_fn(n, _written, total):
+        global progress_save_n
+        progress_save_n += n
+        rate = int(100 * (float(_written) / float(total)))
+        print('\r{0}% '.format(rate))
+
+    copier = client.copier()
+
+    result = copier.copy(oss.CopyObjectRequest(
+            bucket=args.bucket,
+            key=args.key,
+            source_bucket=args.source_bucket,
+            source_key=args.source_key,
+            progress_fn=_progress_fn,
+        ),
+        part_size=300 * 1024 * 1024,
+        parallel_num=5,
+        leave_parts_on_error=True,
+    )
+
+    print(vars(result))
+
+if __name__ == "__main__":
+    main()
+