Support b2id:// URIs in v4 ls and rm commands

b2/_internal/_cli/b2args.py

@@ -17,7 +17,7 @@
 from b2._internal._utils.uri import B2URI, B2URIBase, parse_b2_uri
 
 
-def b2_file_uri(value: str) -> B2URIBase:
+def b2id_or_file_like_b2_uri(value: str) -> B2URIBase:
     b2_uri = parse_b2_uri(value)
     if isinstance(b2_uri, B2URI):
         if b2_uri.is_dir():
@@ -29,39 +29,30 @@ def b2_file_uri(value: str) -> B2URIBase:
     return b2_uri
 
 
-def b2_uri(value: str) -> B2URI:
-    uri = parse_b2_uri(value)
-    if not isinstance(uri, B2URI):
-        raise ValueError(
-            f"B2 URI of the form b2://bucket/path/ is required, but {value} was provided"
-        )
-    return uri
+B2ID_OR_B2_URI_ARG_TYPE = wrap_with_argument_type_error(parse_b2_uri)
+B2ID_OR_FILE_LIKE_B2_URI_ARG_TYPE = wrap_with_argument_type_error(b2id_or_file_like_b2_uri)
 
 
-B2_URI_ARG_TYPE = wrap_with_argument_type_error(b2_uri)
-B2_URI_FILE_ARG_TYPE = wrap_with_argument_type_error(b2_file_uri)
-
-
-def add_b2_uri_argument(parser: argparse.ArgumentParser, name="B2_URI"):
+def add_b2id_or_b2_uri_argument(parser: argparse.ArgumentParser, name="B2_URI"):
     """
-    Add a B2 URI pointing to a bucket, optionally with a directory
-    or a pattern as an argument to the parser.
+    Add B2 URI (b2:// or b2id://) as an argument to the parser.
+    B2 URI can point to a bucket, optionally with a directory or a pattern.
     """
     parser.add_argument(
         name,
-        type=B2_URI_ARG_TYPE,
-        help="B2 URI pointing to a bucket, directory or a pattern, "
+        type=B2ID_OR_B2_URI_ARG_TYPE,
+        help="B2 URI pointing to a bucket, directory, pattern or a file."
         "e.g. b2://yourBucket, b2://yourBucket/file.txt, b2://yourBucket/folder/, "
         "b2://yourBucket/*.txt or b2id://fileId",
-    )
+    ).completer = b2uri_file_completer
 
 
-def add_b2_file_argument(parser: argparse.ArgumentParser, name="B2_URI"):
+def add_b2id_or_file_like_b2_uri_argument(parser: argparse.ArgumentParser, name="B2_URI"):
     """
     Add a B2 URI pointing to a file as an argument to the parser.
     """
     parser.add_argument(
         name,
-        type=B2_URI_FILE_ARG_TYPE,
+        type=B2ID_OR_FILE_LIKE_B2_URI_ARG_TYPE,
         help="B2 URI pointing to a file, e.g. b2://yourBucket/file.txt or b2id://fileId",
     ).completer = b2uri_file_completer

b2/_internal/_utils/uri.py

@@ -19,6 +19,7 @@
     DownloadVersion,
     FileVersion,
 )
+from b2sdk.v2.exception import B2Error
 
 from b2._internal._utils.python_compat import removeprefix, singledispatchmethod
 
@@ -158,3 +159,21 @@ def _(self, uri: B2URI, *args, **kwargs) -> str:
     @get_download_url_by_uri.register
     def _(self, uri: B2FileIdURI, *args, **kwargs) -> str:
         return self.get_download_url_for_fileid(uri.file_id, *args, **kwargs)
+
+    @singledispatchmethod
+    def list_file_versions_by_uri(self, uri, *args, **kwargs):
+        raise NotImplementedError(f"Unsupported URI type: {type(uri)}")
+
+    @list_file_versions_by_uri.register
+    def _(self, uri: B2URI, *args, **kwargs):
+        bucket = self.api.get_bucket_by_name(uri.bucket_name)
+        try:
+            yield from bucket.ls(uri.path, *args, **kwargs)
+        except ValueError as error:
+            # Wrap these errors into B2Error. At the time of writing there's
+            # exactly one – `with_wildcard` being passed without `recursive` option.
+            raise B2Error(error.args[0])
+
+    @list_file_versions_by_uri.register
+    def _(self, uri: B2FileIdURI, *args, **kwargs):
+        yield self.get_file_info_by_uri(uri), None

b2/_internal/console_tool.py

@@ -126,7 +126,10 @@
     autocomplete_install,
 )
 from b2._internal._cli.b2api import _get_b2api_for_profile
-from b2._internal._cli.b2args import add_b2_file_argument, add_b2_uri_argument
+from b2._internal._cli.b2args import (
+    add_b2id_or_b2_uri_argument,
+    add_b2id_or_file_like_b2_uri_argument,
+)
 from b2._internal._cli.const import (
     B2_APPLICATION_KEY_ENV_VAR,
     B2_APPLICATION_KEY_ID_ENV_VAR,
@@ -604,7 +607,7 @@ def _get_file_name_from_args(self, args):
 class B2URIFileArgMixin:
     @classmethod
     def _setup_parser(cls, parser):
-        add_b2_file_argument(parser)
+        add_b2id_or_file_like_b2_uri_argument(parser)
         super()._setup_parser(parser)
 
     def get_b2_uri_from_arg(self, args: argparse.Namespace) -> B2URIBase:
@@ -643,10 +646,10 @@ def get_b2_uri_from_arg(self, args: argparse.Namespace) -> B2URI:
         return B2URI(args.bucketName, args.folderName or '')
 
 
-class B2URIMixin:
+class B2IDOrB2URIMixin:
     @classmethod
     def _setup_parser(cls, parser):
-        add_b2_uri_argument(parser)
+        add_b2id_or_b2_uri_argument(parser)
         super()._setup_parser(parser)
 
     def get_b2_uri_from_arg(self, args: argparse.Namespace) -> B2URI:
@@ -2190,19 +2193,12 @@ def _print_file_version(
 
     def _get_ls_generator(self, args):
         b2_uri = self.get_b2_uri_from_arg(args)
-        bucket = self.api.get_bucket_by_name(b2_uri.bucket_name)
-
-        try:
-            yield from bucket.ls(
-                b2_uri.path,
-                latest_only=not args.versions,
-                recursive=args.recursive,
-                with_wildcard=args.withWildcard,
-            )
-        except ValueError as error:
-            # Wrap these errors into B2Error. At the time of writing there's
-            # exactly one – `with_wildcard` being passed without `recursive` option.
-            raise B2Error(error.args[0])
+        yield from self.api.list_file_versions_by_uri(
+            b2_uri,
+            latest_only=not args.versions,
+            recursive=args.recursive,
+            with_wildcard=args.withWildcard,
+        )
 
     def get_b2_uri_from_arg(self, args: argparse.Namespace) -> B2URI:
         raise NotImplementedError
@@ -2294,7 +2290,7 @@ def format_ls_entry(self, file_version: FileVersion, replication: bool):
         return template % tuple(parameters)
 
 
-class Ls(B2URIMixin, BaseLs):
+class Ls(B2IDOrB2URIMixin, BaseLs):
     """
     {BASELS}
 
@@ -2501,7 +2497,7 @@ def _run(self, args):
         return 1 if failed_on_any_file else 0
 
 
-class Rm(B2URIMixin, BaseRm):
+class Rm(B2IDOrB2URIMixin, BaseRm):
     """
     {BASERM}
 

changelog.d/+ls-rm-b2id-uri.added.md

@@ -0,0 +1 @@
+Add support for deleting a single file by `b2id://` URI in the pre-release version _b2v4.

test/integration/test_b2_command_line.py

@@ -208,6 +208,26 @@ def test_basic(b2_tool, bucket_name, sample_file, tmp_path, b2_uri_args):
     )  # \r? is for Windows, as $ doesn't match \r\n
 
 
+@pytest.mark.cli_version(from_version=4)
+def test_ls_b2id(b2_tool, uploaded_sample_file):
+    b2_tool.should_succeed(
+        ['ls', f"b2id://{uploaded_sample_file['fileId']}"],
+        expected_pattern=f"^{uploaded_sample_file['fileName']}",
+    )
+
+
+@pytest.mark.cli_version(from_version=4)
+def test_rm_b2id(b2_tool, bucket_name, uploaded_sample_file):
+    # remove the file by id
+    b2_tool.should_succeed(['rm', f"b2id://{uploaded_sample_file['fileId']}"])
+
+    # check that the file is gone
+    b2_tool.should_succeed(
+        ['ls', f'b2://{bucket_name}'],
+        expected_pattern='^$',
+    )
+
+
 def test_debug_logs(b2_tool, is_running_on_docker, tmp_path):
     to_be_removed_bucket_name = b2_tool.generate_bucket_name()
     b2_tool.should_succeed(

test/unit/test_console_tool.py

@@ -17,6 +17,7 @@
 from itertools import chain, product
 from typing import List, Optional
 
+import pytest
 from b2sdk import v1
 from b2sdk.v2 import (
     ALL_CAPABILITIES,
@@ -2429,6 +2430,21 @@ def test_passing_api_parameters(self):
             )
             assert parallel_strategy.max_streams == params['--max-download-streams-per-file']
 
+    @pytest.mark.cli_version(from_version=4)
+    def test_ls_b2id(self):
+        self._authorize_account()
+        self._create_my_bucket()
+
+        # Create a file
+        bucket = self.b2_api.get_bucket_by_name('my-bucket')
+        file_version = bucket.upload(UploadSourceBytes(b''), 'test.txt')
+
+        # Condensed output
+        expected_stdout = '''
+                test.txt
+                '''
+        self._run_command(['ls', f'b2id://{file_version.id_}'], expected_stdout, '', 0)
+
 
 class TestConsoleToolWithV1(BaseConsoleToolTest):
     """These tests use v1 interface to perform various setups before running CLI commands"""
@@ -2727,6 +2743,41 @@ def test_rm_skipping_over_errors(self):
         '''
         self._run_command(['ls', '--recursive', *self.b2_uri_args('my-bucket')], expected_stdout)
 
+    @pytest.mark.cli_version(from_version=4)
+    def test_rm_b2id(self):
+        # Create a file
+        file_version = self.bucket.upload(UploadSourceBytes(b''), 'new-file.txt')
+
+        # Before deleting
+        expected_stdout = '''
+        a/test.csv
+        a/test.tsv
+        b/b/test.csv
+        b/b1/test.csv
+        b/b2/test.tsv
+        b/test.txt
+        c/test.csv
+        c/test.tsv
+        new-file.txt
+        '''
+        self._run_command(['ls', '--recursive', 'b2://my-bucket'], expected_stdout)
+
+        # Delete file
+        self._run_command(['rm', '--noProgress', f'b2id://{file_version.id_}'], '', '', 0)
+
+        # After deleting
+        expected_stdout = '''
+        a/test.csv
+        a/test.tsv
+        b/b/test.csv
+        b/b1/test.csv
+        b/b2/test.tsv
+        b/test.txt
+        c/test.csv
+        c/test.tsv
+        '''
+        self._run_command(['ls', '--recursive', 'b2://my-bucket'], expected_stdout)
+
 
 class TestVersionConsoleTool(BaseConsoleToolTest):
     def test_version(self):