Merge pull request #310 from Backblaze/big-copy

Fix big copy

Author: ppolewicz

CHANGELOG.md

@@ -7,17 +7,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Changed
-* Introduce a common thread worker pool for all downloads
 * Don't run coverage in pypy in CI
+* Introduce a common thread worker pool for all downloads
+* Increase http timeout to 20min (for copy using 5GB parts)
 
 ### Added
 * Add pypy-3.8 to test matrix
 * Add support for unverified checksum upload mode
 * Add dedicated exception for unverified email
 * Add a parameter to customize `sync_policy_manager`
+* Add parameters to set the min/max part size for large file upload/copy methods
+* Add CopySourceTooBig exception
+* Add an option to set a custom file version class to FileVersionFactory
 
 ### Fixed
 * Fix downloading files with unverified checksum
+* Fix copying objects larger than 1TB
 
 ## [1.14.1] - 2022-02-23
 

b2sdk/b2http.py

@@ -154,7 +154,7 @@ class B2Http(object):
     """
 
     # timeout for HTTP GET/POST requests
-    TIMEOUT = 900  # 15 minutes as server-side copy can take time
+    TIMEOUT = 1200  # 20 minutes as server-side copy can take time
 
     def __init__(self, api_config: B2HttpApiConfig = DEFAULT_HTTP_API_CONFIG):
         """
@@ -346,6 +346,11 @@ def _translate_errors(cls, fcn, post_params=None):
             if response.status_code not in [200, 206]:
                 # Decode the error object returned by the service
                 error = json.loads(response.content.decode('utf-8')) if response.content else {}
+                extra_error_keys = error.keys() - ('code', 'status', 'message')
+                if extra_error_keys:
+                    logger.debug(
+                        'received error has extra (unsupported) keys: %s', extra_error_keys
+                    )
                 raise interpret_b2_error(
                     int(error.get('status', response.status_code)),
                     error.get('code'),

b2sdk/bucket.py

@@ -13,7 +13,7 @@
 
 from .encryption.setting import EncryptionSetting, EncryptionSettingFactory
 from .encryption.types import EncryptionMode
-from .exception import BucketIdNotFound, FileNotPresent, FileOrBucketNotFound, UnexpectedCloudBehaviour, UnrecognizedBucketType
+from .exception import BucketIdNotFound, CopySourceTooBig, FileNotPresent, FileOrBucketNotFound, UnexpectedCloudBehaviour, UnrecognizedBucketType
 from .file_lock import (
     BucketRetentionSetting,
     FileLockConfiguration,
@@ -550,6 +550,8 @@ def create_file(
         encryption: Optional[EncryptionSetting] = None,
         file_retention: Optional[FileRetentionSetting] = None,
         legal_hold: Optional[LegalHold] = None,
+        min_part_size=None,
+        max_part_size=None,
     ):
         """
         Creates a new file in this bucket using an iterable (list, tuple etc) of remote or local sources.
@@ -574,6 +576,8 @@ def create_file(
         :param b2sdk.v2.EncryptionSetting encryption: encryption settings (``None`` if unknown)
         :param b2sdk.v2.FileRetentionSetting file_retention: file retention setting
         :param bool legal_hold: legal hold setting
+        :param int min_part_size: lower limit of part size for the transfer planner, in bytes
+        :param int max_part_size: upper limit of part size for the transfer planner, in bytes
         """
         return self._create_file(
             self.api.services.emerger.emerge,
@@ -587,6 +591,8 @@ def create_file(
             encryption=encryption,
             file_retention=file_retention,
             legal_hold=legal_hold,
+            min_part_size=min_part_size,
+            max_part_size=max_part_size,
         )
 
     def create_file_stream(
@@ -601,6 +607,8 @@ def create_file_stream(
         encryption: Optional[EncryptionSetting] = None,
         file_retention: Optional[FileRetentionSetting] = None,
         legal_hold: Optional[LegalHold] = None,
+        min_part_size=None,
+        max_part_size=None,
     ):
         """
         Creates a new file in this bucket using a stream of multiple remote or local sources.
@@ -627,6 +635,8 @@ def create_file_stream(
         :param b2sdk.v2.EncryptionSetting encryption: encryption settings (``None`` if unknown)
         :param b2sdk.v2.FileRetentionSetting file_retention: file retention setting
         :param bool legal_hold: legal hold setting
+        :param int min_part_size: lower limit of part size for the transfer planner, in bytes
+        :param int max_part_size: upper limit of part size for the transfer planner, in bytes
         """
         return self._create_file(
             self.api.services.emerger.emerge_stream,
@@ -640,6 +650,8 @@ def create_file_stream(
             encryption=encryption,
             file_retention=file_retention,
             legal_hold=legal_hold,
+            min_part_size=min_part_size,
+            max_part_size=max_part_size,
         )
 
     def _create_file(
@@ -655,6 +667,8 @@ def _create_file(
         encryption: Optional[EncryptionSetting] = None,
         file_retention: Optional[FileRetentionSetting] = None,
         legal_hold: Optional[LegalHold] = None,
+        min_part_size=None,
+        max_part_size=None,
     ):
         validate_b2_file_name(file_name)
         progress_listener = progress_listener or DoNothingProgressListener()
@@ -671,6 +685,8 @@ def _create_file(
             encryption=encryption,
             file_retention=file_retention,
             legal_hold=legal_hold,
+            min_part_size=min_part_size,
+            max_part_size=max_part_size,
         )
 
     def concatenate(
@@ -685,6 +701,8 @@ def concatenate(
         encryption: Optional[EncryptionSetting] = None,
         file_retention: Optional[FileRetentionSetting] = None,
         legal_hold: Optional[LegalHold] = None,
+        min_part_size=None,
+        max_part_size=None,
     ):
         """
         Creates a new file in this bucket by concatenating multiple remote or local sources.
@@ -706,6 +724,8 @@ def concatenate(
         :param b2sdk.v2.EncryptionSetting encryption: encryption settings (``None`` if unknown)
         :param b2sdk.v2.FileRetentionSetting file_retention: file retention setting
         :param bool legal_hold: legal hold setting
+        :param int min_part_size: lower limit of part size for the transfer planner, in bytes
+        :param int max_part_size: upper limit of part size for the transfer planner, in bytes
         """
         return self.create_file(
             WriteIntent.wrap_sources_iterator(outbound_sources),
@@ -718,6 +738,8 @@ def concatenate(
             encryption=encryption,
             file_retention=file_retention,
             legal_hold=legal_hold,
+            min_part_size=min_part_size,
+            max_part_size=max_part_size,
         )
 
     def concatenate_stream(
@@ -806,6 +828,8 @@ def copy(
         source_content_type: Optional[str] = None,
         file_retention: Optional[FileRetentionSetting] = None,
         legal_hold: Optional[LegalHold] = None,
+        min_part_size=None,
+        max_part_size=None,
     ):
         """
         Creates a new file in this bucket by (server-side) copying from an existing file.
@@ -831,6 +855,8 @@ def copy(
         :param str,None source_content_type: source file's content type, useful when copying files with SSE-C
         :param b2sdk.v2.FileRetentionSetting file_retention: file retention setting for the new file.
         :param bool legal_hold: legal hold setting for the new file.
+        :param int min_part_size: lower limit of part size for the transfer planner, in bytes
+        :param int max_part_size: upper limit of part size for the transfer planner, in bytes
         """
 
         copy_source = CopySource(
@@ -844,30 +870,38 @@ def copy(
         if not length:
             # TODO: it feels like this should be checked on lower level - eg. RawApi
             validate_b2_file_name(new_file_name)
-            progress_listener = progress_listener or DoNothingProgressListener()
-            return self.api.services.copy_manager.copy_file(
-                copy_source,
-                new_file_name,
-                content_type=content_type,
-                file_info=file_info,
-                destination_bucket_id=self.id_,
-                progress_listener=progress_listener,
-                destination_encryption=destination_encryption,
-                source_encryption=source_encryption,
-                file_retention=file_retention,
-                legal_hold=legal_hold,
-            ).result()
-        else:
-            return self.create_file(
-                [WriteIntent(copy_source)],
-                new_file_name,
-                content_type=content_type,
-                file_info=file_info,
-                progress_listener=progress_listener,
-                encryption=destination_encryption,
-                file_retention=file_retention,
-                legal_hold=legal_hold,
-            )
+            try:
+                progress_listener = progress_listener or DoNothingProgressListener()
+                return self.api.services.copy_manager.copy_file(
+                    copy_source,
+                    new_file_name,
+                    content_type=content_type,
+                    file_info=file_info,
+                    destination_bucket_id=self.id_,
+                    progress_listener=progress_listener,
+                    destination_encryption=destination_encryption,
+                    source_encryption=source_encryption,
+                    file_retention=file_retention,
+                    legal_hold=legal_hold,
+                ).result()
+            except CopySourceTooBig as e:
+                copy_source.length = e.size
+                progress_listener = DoNothingProgressListener()
+                logger.warning(
+                    'a copy of large object of unknown size is upgraded to the large file interface. No progress report will be provided.'
+                )
+        return self.create_file(
+            [WriteIntent(copy_source)],
+            new_file_name,
+            content_type=content_type,
+            file_info=file_info,
+            progress_listener=progress_listener,
+            encryption=destination_encryption,
+            file_retention=file_retention,
+            legal_hold=legal_hold,
+            min_part_size=min_part_size,
+            max_part_size=max_part_size,
+        )
 
     def delete_file_version(self, file_id, file_name):
         """

b2sdk/exception.py

@@ -19,6 +19,8 @@
     r'^more than one upload using auth token (?P<token>[^)]+)$'
 )
 
+COPY_SOURCE_TOO_BIG_ERROR_MESSAGE_RE = re.compile(r'^Copy source too big: (?P<size>[\d]+)$')
+
 
 class B2Error(Exception, metaclass=ABCMeta):
     def __init__(self, *args, **kwargs):
@@ -312,6 +314,12 @@ def __str__(self):
         return '%s (%s)' % (self.message, self.code)
 
 
+class CopySourceTooBig(BadRequest):
+    def __init__(self, message, code, size: int):
+        super().__init__(message, code)
+        self.size = size
+
+
 class Unauthorized(B2Error):
     def __init__(self, message, code):
         super(Unauthorized, self).__init__()
@@ -543,11 +551,19 @@ def interpret_b2_error(
         return PartSha1Mismatch(post_params.get('fileId'))
     elif status == 400 and code == "bad_bucket_id":
         return BucketIdNotFound(post_params.get('bucketId'))
-    elif status == 400 and code == "bad_request":
+    elif status == 400 and code in ('bad_request', 'auth_token_limit', 'source_too_large'):
+        # it's "bad_request" on 2022-03-29, but will become 'auth_token_limit' in 2022-04  # TODO: cleanup after 2022-05-01
         matcher = UPLOAD_TOKEN_USED_CONCURRENTLY_ERROR_MESSAGE_RE.match(message)
         if matcher is not None:
             token = matcher.group('token')
             return UploadTokenUsedConcurrently(token)
+
+        # it's "bad_request" on 2022-03-29, but will become 'source_too_large' in 2022-04  # TODO: cleanup after 2022-05-01
+        matcher = COPY_SOURCE_TOO_BIG_ERROR_MESSAGE_RE.match(message)
+        if matcher is not None:
+            size = int(matcher.group('size'))
+            return CopySourceTooBig(size)
+
         return BadRequest(message, code)
     elif status == 400:
         return BadRequest(message, code)

b2sdk/file_version.py

@@ -352,6 +352,8 @@ class FileVersionFactory(object):
     Construct :py:class:`b2sdk.v2.FileVersion` objects from api responses.
     """
 
+    FILE_VERSION_CLASS = FileVersion
+
     def __init__(self, api: 'B2Api'):
         self.api = api
 
@@ -408,8 +410,7 @@ def from_api_response(self, file_version_dict, force_action=None):
         file_retention = FileRetentionSetting.from_file_version_dict(file_version_dict)
 
         legal_hold = LegalHold.from_file_version_dict(file_version_dict)
-
-        return FileVersion(
+        return self.FILE_VERSION_CLASS(
             self.api,
             id_,
             file_name,