Replace yapf with ruff

CONTRIBUTING.md

@@ -8,7 +8,7 @@ In order to make it easier to contribute, core developers of this project:
 * maintain a set of unit tests
 * maintain a set of integration tests (run with a production cloud)
 * maintain development automation tools using [nox](https://github.com/theacodes/nox) that can easily:
-  * format the code using [yapf](https://github.com/google/yapf) and [ruff](https://github.com/astral-sh/ruff)
+  * format the code using [ruff](https://github.com/astral-sh/ruff)
   * run linters to find subtle/potential issues with maintainability
   * run the test suite on multiple Python versions using [pytest](https://github.com/pytest-dev/pytest)
 * maintain Continuous Integration (by using GitHub Actions) that:

b2/_internal/_cli/arg_parser_types.py

@@ -15,14 +15,14 @@
 import arrow
 from b2sdk.v2 import RetentionPeriod
 
-_arrow_version = tuple(int(p) for p in arrow.__version__.split("."))
+_arrow_version = tuple(int(p) for p in arrow.__version__.split('.'))
 
 
 def parse_comma_separated_list(s):
     """
     Parse comma-separated list.
     """
-    return [word.strip() for word in s.split(",")]
+    return [word.strip() for word in s.split(',')]
 
 
 def parse_millis_from_float_timestamp(s):
@@ -31,9 +31,9 @@ def parse_millis_from_float_timestamp(s):
     """
     parsed = arrow.get(float(s))
     if _arrow_version < (1, 0, 0):
-        return int(parsed.format("XSSS"))
+        return int(parsed.format('XSSS'))
     else:
-        return int(parsed.format("x")[:13])
+        return int(parsed.format('x')[:13])
 
 
 def parse_range(s):

b2/_internal/_cli/argcompleters.py

@@ -19,11 +19,12 @@ def bucket_name_completer(prefix, parsed_args, **kwargs):
     from b2sdk.v2 import unprintable_to_hex
 
     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_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)
+            (bucket.name, f'b2://{bucket.name}') for bucket in api.list_buckets(use_cache=True)
         )
     ]
     return res
@@ -69,29 +70,29 @@ def b2uri_file_completer(prefix: str, parsed_args, **kwargs):
         prefix_without_scheme = removeprefix(prefix, 'b2://')
         if '/' not in prefix_without_scheme:
             return [
-                f"b2://{unprintable_to_hex(bucket.name)}/"
+                f'b2://{unprintable_to_hex(bucket.name)}/'
                 for bucket in api.list_buckets(use_cache=True)
             ]
 
         b2_uri = parse_b2_uri(prefix)
         bucket = api.get_bucket_by_name(b2_uri.bucket_name)
         file_versions = bucket.ls(
-            f"{b2_uri.path}*",
+            f'{b2_uri.path}*',
             latest_only=True,
             recursive=True,
             fetch_count=LIST_FILE_NAMES_MAX_LIMIT,
             with_wildcard=True,
         )
         return [
-            unprintable_to_hex(f"b2://{bucket.name}/{file_version.file_name}")
+            unprintable_to_hex(f'b2://{bucket.name}/{file_version.file_name}')
             for file_version, folder_name in islice(file_versions, LIST_FILE_NAMES_MAX_LIMIT)
             if file_version
         ]
     elif prefix.startswith('b2id://'):
         # listing all files from all buckets is unreasonably expensive
-        return ["b2id://"]
+        return ['b2id://']
     else:
         return [
-            "b2://",
-            "b2id://",
+            'b2://',
+            'b2id://',
         ]

b2/_internal/_cli/autocomplete_cache.py

@@ -57,13 +57,14 @@ def __init__(self, dir_path: pathlib.Path | None = None) -> None:
 
     def _cache_dir(self) -> pathlib.Path:
         if not self._dir:
-            self._dir = pathlib.Path(
-                platformdirs.user_cache_dir(appname='b2', appauthor='backblaze')
-            ) / 'autocomplete'
+            self._dir = (
+                pathlib.Path(platformdirs.user_cache_dir(appname='b2', appauthor='backblaze'))
+                / 'autocomplete'
+            )
         return self._dir
 
     def _fname(self, identifier: str) -> str:
-        return f"b2-autocomplete-cache-{identifier}.pickle"
+        return f'b2-autocomplete-cache-{identifier}.pickle'
 
     def get_pickle(self, identifier: str) -> bytes | None:
         path = self._cache_dir() / self._fname(identifier)
@@ -93,7 +94,7 @@ def __init__(
         self,
         tracker: StateTracker,
         store: PickleStore,
-        unpickle: Callable[[bytes], argparse.ArgumentParser] | None = None
+        unpickle: Callable[[bytes], argparse.ArgumentParser] | None = None,
     ):
         self._tracker = tracker
         self._store = store

b2/_internal/_cli/autocomplete_install.py

@@ -37,9 +37,9 @@ def autocomplete_install(prog: str, shell: str = 'bash') -> None:
     try:
         autocomplete_installer = SHELL_REGISTRY.get(shell, prog=prog)
     except RegistryKeyError:
-        raise AutocompleteInstallError(f"Unsupported shell: {shell}")
+        raise AutocompleteInstallError(f'Unsupported shell: {shell}')
     autocomplete_installer.install()
-    logger.info("Autocomplete for %s has been enabled.", prog)
+    logger.info('Autocomplete for %s has been enabled.', prog)
 
 
 class ShellAutocompleteInstaller(abc.ABC):
@@ -53,8 +53,9 @@ def install(self) -> None:
         script_path = self.create_script()
         if not self.is_enabled():
             logger.info(
-                "%s completion doesn't seem to be autoloaded from %s.", self.shell_exec,
-                script_path.parent
+                "%s completion doesn't seem to be autoloaded from %s.",
+                self.shell_exec,
+                script_path.parent,
             )
             try:
                 self.force_enable(script_path)
@@ -64,15 +65,15 @@ def install(self) -> None:
                 )
 
             if not self.is_enabled():
-                logger.error("Autocomplete is still not enabled.")
-                raise AutocompleteInstallError(f"Autocomplete for {self.prog} install failed.")
+                logger.error('Autocomplete is still not enabled.')
+                raise AutocompleteInstallError(f'Autocomplete for {self.prog} install failed.')
 
     def create_script(self) -> Path:
         """Create autocomplete for the given program."""
         shellcode = self.get_shellcode()
 
         script_path = self.get_script_path()
-        logger.info("Creating autocompletion script under %s", script_path)
+        logger.info('Creating autocompletion script under %s', script_path)
         script_path.parent.mkdir(exist_ok=True, parents=True, mode=0o755)
         script_path.write_text(shellcode)
         return script_path
@@ -116,25 +117,25 @@ def force_enable(self, completion_script: Path) -> None:
         """Enable autocomplete for the given program, common logic."""
         rc_path = self.get_rc_path()
         if rc_path.exists() and rc_path.read_text().strip():
-            bck_path = rc_path.with_suffix(f".{datetime.now():%Y-%m-%dT%H-%M-%S}.bak")
-            logger.warning("Backing up %s to %s", rc_path, bck_path)
+            bck_path = rc_path.with_suffix(f'.{datetime.now():%Y-%m-%dT%H-%M-%S}.bak')
+            logger.warning('Backing up %s to %s', rc_path, bck_path)
             try:
                 shutil.copyfile(rc_path, bck_path)
             except OSError as e:
                 raise AutocompleteInstallError(
-                    f"Failed to backup {rc_path} under {bck_path}"
+                    f'Failed to backup {rc_path} under {bck_path}'
                 ) from e
-        logger.warning("Explicitly adding %s to %s", completion_script, rc_path)
+        logger.warning('Explicitly adding %s to %s', completion_script, rc_path)
         add_or_update_shell_section(
-            rc_path, f"{self.prog} autocomplete", self.prog, self.get_rc_section(completion_script)
+            rc_path, f'{self.prog} autocomplete', self.prog, self.get_rc_section(completion_script)
         )
 
     def get_rc_section(self, completion_script: Path) -> str:
-        return f"source {quote(str(completion_script))}"
+        return f'source {quote(str(completion_script))}'
 
     def get_script_path(self) -> Path:
         """Get autocomplete script path for the given program, common logic."""
-        script_dir = Path(f"~/.{self.shell_exec}_completion.d/").expanduser()
+        script_dir = Path(f'~/.{self.shell_exec}_completion.d/').expanduser()
         return script_dir / self.prog
 
     def is_enabled(self) -> bool:
@@ -145,13 +146,13 @@ def is_enabled(self) -> bool:
 @SHELL_REGISTRY.register('bash')
 class BashAutocompleteInstaller(BashLikeAutocompleteInstaller):
     shell_exec = 'bash'
-    rc_file_path = "~/.bashrc"
+    rc_file_path = '~/.bashrc'
 
 
 @SHELL_REGISTRY.register('zsh')
 class ZshAutocompleteInstaller(BashLikeAutocompleteInstaller):
     shell_exec = 'zsh'
-    rc_file_path = "~/.zshrc"
+    rc_file_path = '~/.zshrc'
 
     def get_rc_section(self, completion_script: Path) -> str:
         return textwrap.dedent(
@@ -163,7 +164,7 @@ def get_rc_section(self, completion_script: Path) -> str:
 
     def get_script_path(self) -> Path:
         """Custom get_script_path for Zsh, if the structure differs from the base implementation."""
-        return Path("~/.zsh/completion/").expanduser() / f"_{self.prog}"
+        return Path('~/.zsh/completion/').expanduser() / f'_{self.prog}'
 
     def is_enabled(self) -> bool:
         rc_path = self.get_rc_path()
@@ -181,31 +182,32 @@ def is_enabled(self) -> bool:
 @SHELL_REGISTRY.register('fish')
 class FishAutocompleteInstaller(ShellAutocompleteInstaller):
     shell_exec = 'fish'
-    rc_file_path = "~/.config/fish/config.fish"
+    rc_file_path = '~/.config/fish/config.fish'
 
     def force_enable(self, completion_script: Path) -> None:
         raise NotImplementedError("Fish shell doesn't support manual completion enabling.")
 
     def get_script_path(self) -> Path:
         """Get autocomplete script path for the given program, common logic."""
         complete_paths = [
-            Path(p) for p in shlex.split(
+            Path(p)
+            for p in shlex.split(
                 subprocess.run(
                     [self.shell_exec, '-c', 'echo $fish_complete_path'],
                     timeout=30,
                     text=True,
                     check=True,
-                    capture_output=True
+                    capture_output=True,
                 ).stdout
             )
         ]
-        user_path = Path("~/.config/fish/completions").expanduser()
+        user_path = Path('~/.config/fish/completions').expanduser()
         if complete_paths:
             target_path = user_path if user_path in complete_paths else complete_paths[0]
         else:
-            logger.warning("$fish_complete_path is empty, falling back to %r", user_path)
+            logger.warning('$fish_complete_path is empty, falling back to %r', user_path)
             target_path = user_path
-        return target_path / f"{self.prog}.fish"
+        return target_path / f'{self.prog}.fish'
 
     def is_enabled(self) -> bool:
         """
@@ -216,11 +218,13 @@ def is_enabled(self) -> bool:
         named filenames).
         """
         environ = os.environ.copy()
-        environ.setdefault("TERM", "xterm")  # TERM has to be set for fish to load completions
+        environ.setdefault('TERM', 'xterm')  # TERM has to be set for fish to load completions
         return _silent_success_run_with_tty(
             [
-                self.shell_exec, '-i', '-c',
-                f'string length -q -- (complete -C{quote(f"{self.prog} ")} >/dev/null && complete -c {quote(self.prog)})'
+                self.shell_exec,
+                '-i',
+                '-c',
+                f'string length -q -- (complete -C{quote(f"{self.prog} ")} >/dev/null && complete -c {quote(self.prog)})',
             ],
             env=environ,
         )
@@ -233,8 +237,8 @@ def _silent_success_run_with_tty(
     if emulate_tty and not find_spec('pexpect'):
         emulate_tty = False
         logger.warning(
-            "pexpect is needed to check autocomplete installation correctness without tty. "
-            "You can install it via `pip install pexpect`."
+            'pexpect is needed to check autocomplete installation correctness without tty. '
+            'You can install it via `pip install pexpect`.'
         )
     run_func = _silent_success_run_with_pty if emulate_tty else _silent_success_run
     return run_func(cmd, timeout=timeout, env=env)
@@ -255,12 +259,15 @@ def _silent_success_run(cmd: list[str], timeout: int = 30, env: dict | None = No
     except subprocess.TimeoutExpired:
         p.kill()
         stdout, stderr = p.communicate(timeout=1)
-        logger.warning("Command %r timed out, stdout: %r, stderr: %r", cmd, stdout, stderr)
+        logger.warning('Command %r timed out, stdout: %r, stderr: %r', cmd, stdout, stderr)
     else:
         logger.log(
             logging.DEBUG if p.returncode == 0 else logging.WARNING,
-            "Command %r exited with code %r, stdout: %r, stderr: %r", cmd, p.returncode, stdout,
-            stderr
+            'Command %r exited with code %r, stdout: %r, stderr: %r',
+            cmd,
+            p.returncode,
+            stdout,
+            stderr,
         )
     return p.returncode == 0
 
@@ -281,30 +288,33 @@ def _silent_success_run_with_pty(
         child.logfile_read = output
         child.expect(pexpect.EOF)
     except pexpect.TIMEOUT:
-        logger.warning("Command %r timed out, output: %r", cmd, output.getvalue())
+        logger.warning('Command %r timed out, output: %r', cmd, output.getvalue())
         child.kill(signal.SIGKILL)
         return False
     finally:
         child.close()
 
     logger.log(
         logging.DEBUG if child.exitstatus == 0 else logging.WARNING,
-        "Command %r exited with code %r, output: %r", cmd, child.exitstatus, output.getvalue()
+        'Command %r exited with code %r, output: %r',
+        cmd,
+        child.exitstatus,
+        output.getvalue(),
     )
     return child.exitstatus == 0
 
 
 def add_or_update_shell_section(
-    path: Path, section: str, managed_by: str, content: str, comment_sign="#"
+    path: Path, section: str, managed_by: str, content: str, comment_sign='#'
 ) -> None:
     """Add or update a section in a file."""
-    section_start = f"{comment_sign} >>> {section} >>>"
-    section_end = f"{comment_sign} <<< {section} <<<"
+    section_start = f'{comment_sign} >>> {section} >>>'
+    section_end = f'{comment_sign} <<< {section} <<<'
     assert section_end not in content
     try:
         file_content = path.read_text()
     except FileNotFoundError:
-        file_content = ""
+        file_content = ''
 
     full_content = f"""
 {section_start}
@@ -319,7 +329,7 @@ def add_or_update_shell_section(
     if pattern.search(file_content):
         file_content = pattern.sub(full_content, file_content)
     else:
-        file_content += f"\n{full_content}\n"
+        file_content += f'\n{full_content}\n'
     path.write_text(file_content)
 
 

b2/_internal/_cli/b2api.py

@@ -29,7 +29,6 @@ def _get_b2api_for_profile(
     raise_if_does_not_exist: bool = False,
     **kwargs,
 ) -> B2Api:
-
     if raise_if_does_not_exist:
         account_info_file = SqliteAccountInfo.get_user_account_info_path(profile=profile)
         if not os.path.exists(account_info_file):
@@ -64,4 +63,6 @@ def _get_inmemory_b2api(**kwargs) -> B2Api:
 
 
 def _get_b2httpapiconfig():
-    return B2HttpApiConfig(user_agent_append=os.environ.get(B2_USER_AGENT_APPEND_ENV_VAR),)
+    return B2HttpApiConfig(
+        user_agent_append=os.environ.get(B2_USER_AGENT_APPEND_ENV_VAR),
+    )