add b2 file server-side-copy command

b2/_internal/_cli/b2args.py

@@ -104,7 +104,6 @@ def b2id_or_file_like_b2_uri_or_bucket_name(value: str, *,
 B2ID_OR_B2_URI_OR_ALL_BUCKETS_ARG_TYPE = wrap_with_argument_type_error(
     functools.partial(parse_b2_uri, allow_all_buckets=True)
 )
-B2ID_OR_FILE_LIKE_B2_URI_ARG_TYPE = wrap_with_argument_type_error(b2id_or_file_like_b2_uri)
 
 
 def add_bucket_name_argument(
@@ -204,13 +203,17 @@ def add_b2id_or_b2_bucket_uri_argument(parser: argparse.ArgumentParser, name="B2
     return arg
 
 
-def add_b2id_or_file_like_b2_uri_argument(parser: argparse.ArgumentParser, name="B2_URI"):
+def add_b2id_or_file_like_b2_uri_argument(
+    parser: argparse.ArgumentParser, name="B2_URI", *, by_id: Optional[bool] = None
+):
     """
     Add a B2 URI pointing to a file as an argument to the parser.
     """
     arg = parser.add_argument(
         name,
-        type=B2ID_OR_FILE_LIKE_B2_URI_ARG_TYPE,
+        type=wrap_with_argument_type_error(
+            functools.partial(b2id_or_file_like_b2_uri, by_id=by_id)
+        ),
         help="B2 URI pointing to a file, e.g. b2://yourBucket/file.txt or b2id://fileId",
     )
     arg.completer = b2uri_file_completer

b2/_internal/_utils/uri.py

@@ -211,3 +211,17 @@ def _(self, uri: B2URI, *args, filters: Sequence[Filter] = (), **kwargs):
     @ls.register
     def _(self, uri: B2FileIdURI, *args, **kwargs):
         yield self.get_file_info_by_uri(uri), None
+
+    @singledispatchmethod
+    def copy_by_uri(self, uri, *args, **kwargs):
+        raise NotImplementedError(f"Unsupported URI type: {type(uri)}")
+
+    @copy_by_uri.register
+    def _(self, source: B2FileIdURI, destination: B2URI, *args, **kwargs):
+        destination_bucket = self.get_bucket_by_name(destination.bucket_name)
+        return destination_bucket.copy(source.file_id, destination.path, *args, **kwargs)
+
+    @copy_by_uri.register
+    def _(self, source: B2URI, destination: B2URI, *args, **kwargs):
+        file_info = self.get_file_info_by_uri(source)
+        return self.copy_by_uri(B2FileIdURI(file_info.id_), destination, *args, **kwargs)

b2/_internal/console_tool.py

@@ -1466,7 +1466,7 @@ def _run(self, args):
         return 0
 
 
-class FileCopyByIdBase(
+class FileServerSideCopyBase(
     HeaderFlagsMixin, DestinationSseMixin, SourceSseMixin, FileRetentionSettingMixin,
     LegalHoldMixin, Command
 ):
@@ -1520,12 +1520,14 @@ def _setup_parser(cls, parser):
         add_normalized_argument(info_group, '--info', action='append')
         add_normalized_argument(info_group, '--no-info', action='store_true', default=False)
 
-        parser.add_argument('sourceFileId')
-        parser.add_argument('destinationBucketName')
-        parser.add_argument('b2FileName')
-
         super()._setup_parser(parser)  # add parameters from the mixins
 
+    def get_source_b2_uri(self, args) -> B2URIBase:
+        raise NotImplementedError
+
+    def get_destination_b2_uri(self, args) -> B2URI:
+        raise NotImplementedError
+
     def _run(self, args):
         file_infos = None
         if args.info:
@@ -1540,7 +1542,9 @@ def _run(self, args):
                 '--content-type and --info.'
             )
 
-        bucket = self.api.get_bucket_by_name(args.destinationBucketName)
+        source_b2_uri = self.get_source_b2_uri(args)
+        destination_b2_uri = self.get_destination_b2_uri(args)
+
         destination_encryption_setting = self._get_destination_sse_setting(args)
         source_encryption_setting = self._get_source_sse_setting(args)
         legal_hold = self._get_legal_hold_setting(args)
@@ -1553,16 +1557,16 @@ def _run(self, args):
         else:
             range_args = {}
         source_file_info, source_content_type = self._determine_source_metadata(
-            source_file_id=args.sourceFileId,
+            source_b2_uri=source_b2_uri,
             source_encryption=source_encryption_setting,
             destination_encryption=destination_encryption_setting,
             target_content_type=args.content_type,
             target_file_info=file_infos,
             fetch_if_necessary=args.fetch_metadata,
         )
-        file_version = bucket.copy(
-            args.sourceFileId,
-            args.b2FileName,
+        file_version = self.api.copy_by_uri(
+            source_b2_uri,
+            destination_b2_uri,
             **range_args,
             content_type=args.content_type,
             file_info=file_infos,
@@ -1583,7 +1587,7 @@ def _is_ssec(self, encryption: EncryptionSetting | None):
 
     def _determine_source_metadata(
         self,
-        source_file_id: str,
+        source_b2_uri: B2URIBase,
         destination_encryption: EncryptionSetting | None,
         source_encryption: EncryptionSetting | None,
         target_file_info: dict | None,
@@ -1602,10 +1606,25 @@ def _determine_source_metadata(
                 'Attempting to copy file with metadata while either source or destination uses '
                 'SSE-C. Use --fetch-metadata to fetch source file metadata before copying.'
             )
-        source_file_version = self.api.get_file_info(source_file_id)
+        source_file_version = self.api.get_file_info(source_b2_uri)
         return source_file_version.file_info, source_file_version.content_type
 
 
+class FileServerSideCopyLegacyBase(FileServerSideCopyBase):
+    @classmethod
+    def _setup_parser(cls, parser):
+        parser.add_argument('sourceFileId')
+        parser.add_argument('destinationBucketName')
+        parser.add_argument('b2FileName')
+        super()._setup_parser(parser)
+
+    def get_source_b2_uri(self, args) -> B2URIBase:
+        return B2FileIdURI(args.sourceFileId)
+
+    def get_destination_b2_uri(self, args) -> B2URI:
+        return B2URI(args.destinationBucketName, args.b2FileName)
+
+
 class BucketCreateBase(DefaultSseMixin, LifecycleRulesMixin, Command):
     """
     Create a new bucket.
@@ -5094,10 +5113,10 @@ class File(Command):
     .. code-block::
 
         {NAME} file cat b2://yourBucket/file.txt
-        {NAME} file copy-by-id sourceFileId yourBucket file.txt
         {NAME} file download b2://yourBucket/file.txt localFile.txt
         {NAME} file hide b2://yourBucket/file.txt
         {NAME} file info b2://yourBucket/file.txt
+        {NAME} file server-side-copy b2://yourBucket/file.txt b2://otherBucket/file2.txt
         {NAME} file update --legal-hold off b2://yourBucket/file.txt
         {NAME} file upload yourBucket localFile.txt file.txt
         {NAME} file url b2://yourBucket/file.txt
@@ -5136,9 +5155,28 @@ class FileDownload(B2URIFileArgMixin, FileDownloadBase):
 
 
 @File.subcommands_registry.register
-class FileCopyById(FileCopyByIdBase):
-    __doc__ = FileCopyByIdBase.__doc__
+class FileServerSideCopy(FileServerSideCopyBase):
+    __doc__ = FileServerSideCopyBase.__doc__
+    COMMAND_NAME = 'server-side-copy'
+
+    @classmethod
+    def _setup_parser(cls, parser):
+        add_b2id_or_file_like_b2_uri_argument(parser, "sourceB2Uri")
+        add_b2id_or_file_like_b2_uri_argument(parser, "destinationB2Uri", by_id=False)
+        super()._setup_parser(parser)
+
+    def get_source_b2_uri(self, args) -> B2URIBase:
+        return args.sourceB2Uri
+
+    def get_destination_b2_uri(self, args) -> B2URI:
+        return args.destinationB2Uri
+
+
+@File.subcommands_registry.register
+class FileCopyById(CmdReplacedByMixin, FileServerSideCopyLegacyBase):
+    __doc__ = FileServerSideCopyBase.__doc__
     COMMAND_NAME = 'copy-by-id'
+    replaced_by_cmd = (File, FileServerSideCopy)
 
 
 @File.subcommands_registry.register
@@ -5216,9 +5254,9 @@ class DownloadFileByName(CmdReplacedByMixin, B2URIBucketNFilenameArgMixin, FileD
     replaced_by_cmd = (File, FileDownload)
 
 
-class CopyFileById(CmdReplacedByMixin, FileCopyByIdBase):
-    __doc__ = FileCopyByIdBase.__doc__
-    replaced_by_cmd = (File, FileCopyById)
+class CopyFileById(CmdReplacedByMixin, FileServerSideCopyLegacyBase):
+    __doc__ = FileServerSideCopyBase.__doc__
+    replaced_by_cmd = (File, FileServerSideCopy)
 
 
 class HideFile(CmdReplacedByMixin, HideFileBase):

changelog.d/+add_file_server_side_copy.added.md

@@ -0,0 +1,2 @@
+Add `b2 file server-side-copy b2id:/XXX` (also accepts `b2://bucket/objectName` syntax).
+Add deprecation notice to `b2 file copy-by-id` - use `b2 file server-side-copy` instead in new scripts.