Merge pull request #1015 from reef-technologies/notifications

Event Notifications rules CLI

.github/workflows/cd.yml

@@ -25,7 +25,7 @@ jobs:
         run: |
           export IS_PRERELEASE=$([[ ${{ github.ref }} =~ [^0-9]$ ]] && echo true || echo false)
           echo "prerelease=$IS_PRERELEASE" >> $GITHUB_OUTPUT
-          export PUBLISH_DOCKER=$([[ $IS_PRERELEASE == 'false' && ${{ secrets.DOCKERHUB_USERNAME }} != '' ]] && echo true || echo false) 
+          export PUBLISH_DOCKER=$([[ $IS_PRERELEASE == 'false' && "${{ secrets.DOCKERHUB_USERNAME }}" != '' ]] && echo true || echo false)
           echo "publish_docker=$PUBLISH_DOCKER" >> $GITHUB_OUTPUT
       - uses: actions/checkout@v4
         with:

b2/_internal/_b2v4/registry.py

@@ -55,3 +55,4 @@
 B2.register_subcommand(Version)
 B2.register_subcommand(License)
 B2.register_subcommand(InstallAutocomplete)
+B2.register_subcommand(NotificationRules)

b2/_internal/_cli/argcompleters.py

@@ -7,11 +7,11 @@
 # License https://www.backblaze.com/using_b2_code.html
 #
 ######################################################################
+import itertools
 
 # We import all the necessary modules lazily in completers in order
 # to avoid upfront cost of the imports when argcompleter is used for
 # autocompletions.
-
 from itertools import islice
 
 
@@ -20,7 +20,12 @@ def bucket_name_completer(prefix, parsed_args, **kwargs):
 
     from b2._internal._cli.b2api import _get_b2api_for_profile
     api = _get_b2api_for_profile(getattr(parsed_args, 'profile', None))
-    res = [unprintable_to_hex(bucket.name) for bucket in api.list_buckets(use_cache=True)]
+    res = [
+        unprintable_to_hex(bucket_name_alias)
+        for bucket_name_alias in itertools.chain.from_iterable(
+            (bucket.name, f"b2://{bucket.name}") for bucket in api.list_buckets(use_cache=True)
+        )
+    ]
     return res
 
 

b2/_internal/_cli/b2args.py

@@ -16,12 +16,12 @@
 from typing import Optional, Tuple
 
 from b2._internal._cli.arg_parser_types import wrap_with_argument_type_error
-from b2._internal._cli.argcompleters import b2uri_file_completer
+from b2._internal._cli.argcompleters import b2uri_file_completer, bucket_name_completer
 from b2._internal._cli.const import (
     B2_APPLICATION_KEY_ENV_VAR,
     B2_APPLICATION_KEY_ID_ENV_VAR,
 )
-from b2._internal._utils.uri import B2URI, B2URIBase, parse_b2_uri
+from b2._internal._utils.uri import B2URI, B2URIBase, parse_b2_uri, parse_uri
 
 
 def b2id_or_file_like_b2_uri(value: str) -> B2URIBase:
@@ -36,13 +36,54 @@ def b2id_or_file_like_b2_uri(value: str) -> B2URIBase:
     return b2_uri
 
 
+def parse_bucket_name(value: str, allow_all_buckets: bool = False) -> str:
+    uri = parse_uri(value, allow_all_buckets=allow_all_buckets)
+    if isinstance(uri, B2URI):
+        if uri.path:
+            raise ValueError(
+                f"Expected a bucket name, but {value!r} was provided which contains path part: {uri.path!r}"
+            )
+        return uri.bucket_name
+    return str(value)
+
+
 B2ID_OR_B2_URI_ARG_TYPE = wrap_with_argument_type_error(parse_b2_uri)
 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(
+    parser: argparse.ArgumentParser, name="bucketName", help="Target bucket name", nargs=None
+):
+    parser.add_argument(
+        name,
+        type=wrap_with_argument_type_error(
+            functools.partial(parse_bucket_name, allow_all_buckets=nargs == "?")
+        ),
+        help=help,
+        nargs=nargs
+    ).completer = bucket_name_completer
+
+
+def add_b2_uri_argument(
+    parser: argparse.ArgumentParser,
+    name="B2_URI",
+    help="B2 URI pointing to a bucket with optional path, e.g. b2://yourBucket, b2://yourBucket/file.txt, b2://yourBucket/folderName/",
+):
+    """
+    Add B2 URI as an argument to the parser.
+
+    B2 URI can point to a bucket optionally with a object name prefix (directory).
+    """
+    parser.add_argument(
+        name,
+        type=wrap_with_argument_type_error(functools.partial(parse_b2_uri, allow_b2id=False)),
+        help=help,
+    ).completer = b2uri_file_completer
+
+
 def add_b2id_or_b2_uri_argument(
     parser: argparse.ArgumentParser, name="B2_URI", *, allow_all_buckets: bool = False
 ):

b2/_internal/_cli/obj_dumps.py

@@ -0,0 +1,73 @@
+######################################################################
+#
+# File: b2/_internal/_cli/obj_dumps.py
+#
+# Copyright 2024 Backblaze Inc. All Rights Reserved.
+#
+# License https://www.backblaze.com/using_b2_code.html
+#
+######################################################################
+import io
+
+from b2sdk.v2 import (
+    unprintable_to_hex,
+)
+
+_simple_repr_map = {
+    False: "false",
+    None: "null",
+    True: "true",
+}
+_simple_repr_map_values = set(_simple_repr_map.values()) | {"yes", "no"}
+
+
+def _yaml_simple_repr(obj):
+    """
+    Like YAML for simple types, but also escapes control characters for safety.
+    """
+    if isinstance(obj, (int, float)) and not isinstance(obj, bool):
+        return str(obj)
+    simple_repr = _simple_repr_map.get(obj)
+    if simple_repr:
+        return simple_repr
+    obj_repr = unprintable_to_hex(str(obj))
+    if isinstance(
+        obj, str
+    ) and (obj == "" or obj_repr.lower() in _simple_repr_map_values or obj_repr.isdigit()):
+        obj_repr = repr(obj)  # add quotes to distinguish from numbers and booleans
+    return obj_repr
+
+
+def _id_name_first_key(item):
+    try:
+        return ("id", "name").index(str(item[0]).lower()), item[0], item[1]
+    except ValueError:
+        return 2, item[0], item[1]
+
+
+def _dump(data, indent=0, skip=False, *, output):
+    prefix = " " * indent
+    if isinstance(data, dict):
+        for idx, (key, value) in enumerate(sorted(data.items(), key=_id_name_first_key)):
+            output.write(f"{'' if skip and idx == 0 else prefix}{_yaml_simple_repr(key)}: ")
+            if isinstance(value, (dict, list)):
+                output.write("\n")
+                _dump(value, indent + 2, output=output)
+            else:
+                _dump(value, 0, True, output=output)
+    elif isinstance(data, list):
+        for idx, item in enumerate(data):
+            output.write(f"{'' if skip and idx == 0 else prefix}- ")
+            _dump(item, indent + 2, True, output=output)
+    else:
+        output.write(f"{'' if skip else prefix}{_yaml_simple_repr(data)}\n")
+
+
+def readable_yaml_dump(data, output: io.TextIOBase) -> None:
+    """
+    Print YAML-like human-readable representation of the data.
+
+    :param data: The data to be printed. Can be a list, dict, or any basic datatype.
+    :param output: An output stream derived from io.TextIOBase where the data is to be printed.
+    """
+    _dump(data, output=output)

b2/_internal/_cli/obj_loads.py

@@ -53,5 +53,8 @@ def validated_loads(data: str, expected_type: type[T] | None = None) -> T:
                 f'Invalid value inputted, expected {describe_type(expected_type)}, got {data!r}, more detail below:\n{errors}'
             ) from e
     else:
-        val = json.loads(data)
+        try:
+            val = json.loads(data)
+        except json.JSONDecodeError as e:
+            raise argparse.ArgumentTypeError(f'{data!r} is not a valid JSON value') from e
     return val

b2/_internal/_utils/uri.py

@@ -97,21 +97,28 @@ def parse_uri(uri: str, *, allow_all_buckets: bool = False) -> Path | B2URI | B2
     return _parse_b2_uri(uri, parsed, allow_all_buckets=allow_all_buckets)
 
 
-def parse_b2_uri(uri: str, *, allow_all_buckets: bool = False) -> B2URI | B2FileIdURI:
+def parse_b2_uri(
+    uri: str, *, allow_all_buckets: bool = False, allow_b2id: bool = True
+) -> B2URI | B2FileIdURI:
     """
     Parse B2 URI.
 
     :param uri: string to parse
     :param allow_all_buckets: if True, allow `b2://` without a bucket name to refer to all buckets
+    :param allow_b2id: if True, allow `b2id://` to refer to a file by its id
     :return: B2 URI
     :raises ValueError: if the URI is invalid
     """
     parsed = urllib.parse.urlsplit(uri)
-    return _parse_b2_uri(uri, parsed, allow_all_buckets=allow_all_buckets)
+    return _parse_b2_uri(uri, parsed, allow_all_buckets=allow_all_buckets, allow_b2id=allow_b2id)
 
 
 def _parse_b2_uri(
-    uri, parsed: urllib.parse.SplitResult, allow_all_buckets: bool = False
+    uri,
+    parsed: urllib.parse.SplitResult,
+    *,
+    allow_all_buckets: bool = False,
+    allow_b2id: bool = True
 ) -> B2URI | B2FileIdURI:
     if parsed.scheme in ("b2", "b2id"):
         path = urllib.parse.urlunsplit(parsed._replace(scheme="", netloc=""))
@@ -130,7 +137,7 @@ def _parse_b2_uri(
 
         if parsed.scheme == "b2":
             return B2URI(bucket_name=parsed.netloc, path=removeprefix(path, "/"))
-        elif parsed.scheme == "b2id":
+        elif parsed.scheme == "b2id" and allow_b2id:
             return B2FileIdURI(file_id=parsed.netloc)
     else:
         raise ValueError(f"Unsupported URI scheme: {parsed.scheme!r}")

b2/_internal/b2v3/registry.py

@@ -135,3 +135,4 @@ class Ls(B2URIBucketNFolderNameArgMixin, BaseLs):
 B2.register_subcommand(Version)
 B2.register_subcommand(License)
 B2.register_subcommand(InstallAutocomplete)
+B2.register_subcommand(NotificationRules)