Remove decargs and other references to Python 2 (#5859)

This PR modernizes the codebase by removing Python 2 compatibility code
and simplifying several areas:

- Deleted `BytesQuery` class (replaced with `PathQuery`)
- Removed `decargs()` function that was a no-op in Python 3
- Simplified `print_()` function signature and implementation
- Removed coding-related workarounds in various modules

Author: snejus

beets/dbcore/query.py

@@ -411,39 +411,6 @@ def __init__(
         super().__init__(field_name, pattern_int, fast)
 
 
-class BytesQuery(FieldQuery[bytes]):
-    """Match a raw bytes field (i.e., a path). This is a necessary hack
-    to work around the `sqlite3` module's desire to treat `bytes` and
-    `unicode` equivalently in Python 2. Always use this query instead of
-    `MatchQuery` when matching on BLOB values.
-    """
-
-    def __init__(self, field_name: str, pattern: bytes | str | memoryview):
-        # Use a buffer/memoryview representation of the pattern for SQLite
-        # matching. This instructs SQLite to treat the blob as binary
-        # rather than encoded Unicode.
-        if isinstance(pattern, (str, bytes)):
-            if isinstance(pattern, str):
-                bytes_pattern = pattern.encode("utf-8")
-            else:
-                bytes_pattern = pattern
-            self.buf_pattern = memoryview(bytes_pattern)
-        elif isinstance(pattern, memoryview):
-            self.buf_pattern = pattern
-            bytes_pattern = bytes(pattern)
-        else:
-            raise ValueError("pattern must be bytes, str, or memoryview")
-
-        super().__init__(field_name, bytes_pattern)
-
-    def col_clause(self) -> tuple[str, Sequence[SQLiteType]]:
-        return self.field + " = ?", [self.buf_pattern]
-
-    @classmethod
-    def value_match(cls, pattern: bytes, value: Any) -> bool:
-        return pattern == value
-
-
 class NumericQuery(FieldQuery[str]):
     """Matches numeric fields. A syntax using Ruby-style range ellipses
     (``..``) lets users specify one- or two-sided ranges. For example,

beets/importer/tasks.py

@@ -26,7 +26,8 @@
 
 import mediafile
 
-from beets import autotag, config, dbcore, library, plugins, util
+from beets import autotag, config, library, plugins, util
+from beets.dbcore.query import PathQuery
 
 from .state import ImportState
 
@@ -520,9 +521,7 @@ def record_replaced(self, lib: library.Library):
         )
         replaced_album_ids = set()
         for item in self.imported_items():
-            dup_items = list(
-                lib.items(query=dbcore.query.BytesQuery("path", item.path))
-            )
+            dup_items = list(lib.items(query=PathQuery("path", item.path)))
             self.replaced_items[item] = dup_items
             for dup_item in dup_items:
                 if (

beets/ui/__init__.py

@@ -104,30 +104,15 @@ def _stream_encoding(stream, default="utf-8"):
     return stream.encoding or default
 
 
-def decargs(arglist):
-    """Given a list of command-line argument bytestrings, attempts to
-    decode them to Unicode strings when running under Python 2.
-    """
-    return arglist
-
-
-def print_(*strings, **kwargs):
+def print_(*strings: str, end: str = "\n") -> None:
     """Like print, but rather than raising an error when a character
     is not in the terminal's encoding's character set, just silently
     replaces it.
 
-    The arguments must be Unicode strings: `unicode` on Python 2; `str` on
-    Python 3.
-
     The `end` keyword argument behaves similarly to the built-in `print`
     (it defaults to a newline).
     """
-    if not strings:
-        strings = [""]
-    assert isinstance(strings[0], str)
-
-    txt = " ".join(strings)
-    txt += kwargs.get("end", "\n")
+    txt = " ".join(strings or ("",)) + end
 
     # Encode the string and write it to stdout.
     # On Python 3, sys.stdout expects text strings and uses the
@@ -1308,14 +1293,9 @@ def _set_format(
             setattr(parser.values, option.dest, True)
 
         # Use the explicitly specified format, or the string from the option.
-        if fmt:
-            value = fmt
-        elif value:
-            (value,) = decargs([value])
-        else:
-            value = ""
-
+        value = fmt or value or ""
         parser.values.format = value
+
         if target:
             config[target._format_config_key].set(value)
         else:

beets/ui/commands.py

@@ -28,7 +28,6 @@
 from beets import autotag, config, importer, library, logging, plugins, ui, util
 from beets.autotag import Recommendation, hooks
 from beets.ui import (
-    decargs,
     input_,
     print_,
     print_column_layout,
@@ -1303,7 +1302,7 @@ def _get_choices(self, task):
 # The import command.
 
 
-def import_files(lib, paths, query):
+def import_files(lib, paths: list[bytes], query):
     """Import the files in the given list of paths or matching the
     query.
     """
@@ -1334,7 +1333,7 @@ def import_files(lib, paths, query):
     plugins.send("import", lib=lib, paths=paths)
 
 
-def import_func(lib, opts, args):
+def import_func(lib, opts, args: list[str]):
     config["import"].set_args(opts)
 
     # Special case: --copy flag suppresses import_move (which would
@@ -1343,7 +1342,7 @@ def import_func(lib, opts, args):
         config["import"]["move"] = False
 
     if opts.library:
-        query = decargs(args)
+        query = args
         paths = []
     else:
         query = None
@@ -1356,15 +1355,11 @@ def import_func(lib, opts, args):
         if not paths and not paths_from_logfiles:
             raise ui.UserError("no path specified")
 
-        # On Python 2, we used to get filenames as raw bytes, which is
-        # what we need. On Python 3, we need to undo the "helpful"
-        # conversion to Unicode strings to get the real bytestring
-        # filename.
-        paths = [os.fsencode(p) for p in paths]
+        byte_paths = [os.fsencode(p) for p in paths]
         paths_from_logfiles = [os.fsencode(p) for p in paths_from_logfiles]
 
         # Check the user-specified directories.
-        for path in paths:
+        for path in byte_paths:
             if not os.path.exists(syspath(normpath(path))):
                 raise ui.UserError(
                     "no such file or directory: {}".format(
@@ -1385,14 +1380,14 @@ def import_func(lib, opts, args):
                 )
                 continue
 
-            paths.append(path)
+            byte_paths.append(path)
 
         # If all paths were read from a logfile, and none of them exist, throw
         # an error
         if not paths:
             raise ui.UserError("none of the paths are importable")
 
-    import_files(lib, paths, query)
+    import_files(lib, byte_paths, query)
 
 
 import_cmd = ui.Subcommand(
@@ -1596,7 +1591,7 @@ def list_items(lib, query, album, fmt=""):
 
 
 def list_func(lib, opts, args):
-    list_items(lib, decargs(args), opts.album)
+    list_items(lib, args, opts.album)
 
 
 list_cmd = ui.Subcommand("list", help="query the library", aliases=("ls",))
@@ -1739,7 +1734,7 @@ def update_func(lib, opts, args):
             return
     update_items(
         lib,
-        decargs(args),
+        args,
         opts.album,
         ui.should_move(opts.move),
         opts.pretend,
@@ -1861,7 +1856,7 @@ def fmt_album(a):
 
 
 def remove_func(lib, opts, args):
-    remove_items(lib, decargs(args), opts.album, opts.delete, opts.force)
+    remove_items(lib, args, opts.album, opts.delete, opts.force)
 
 
 remove_cmd = ui.Subcommand(
@@ -1931,7 +1926,7 @@ def show_stats(lib, query, exact):
 
 
 def stats_func(lib, opts, args):
-    show_stats(lib, decargs(args), opts.exact)
+    show_stats(lib, args, opts.exact)
 
 
 stats_cmd = ui.Subcommand(
@@ -2059,7 +2054,7 @@ def modify_parse_args(args):
 
 
 def modify_func(lib, opts, args):
-    query, mods, dels = modify_parse_args(decargs(args))
+    query, mods, dels = modify_parse_args(args)
     if not mods and not dels:
         raise ui.UserError("no modifications specified")
     modify_items(
@@ -2217,7 +2212,7 @@ def move_func(lib, opts, args):
     move_items(
         lib,
         dest,
-        decargs(args),
+        args,
         opts.copy,
         opts.album,
         opts.pretend,
@@ -2298,7 +2293,7 @@ def write_items(lib, query, pretend, force):
 
 
 def write_func(lib, opts, args):
-    write_items(lib, decargs(args), opts.pretend, opts.force)
+    write_items(lib, args, opts.pretend, opts.force)
 
 
 write_cmd = ui.Subcommand("write", help="write tag information to files")

beets/util/__init__.py

@@ -28,6 +28,7 @@
 import tempfile
 import traceback
 from collections import Counter
+from collections.abc import Sequence
 from contextlib import suppress
 from enum import Enum
 from functools import cache
@@ -41,7 +42,6 @@
     AnyStr,
     Callable,
     Generic,
-    Iterable,
     NamedTuple,
     TypeVar,
     Union,
@@ -53,23 +53,17 @@
 from beets.util import hidden
 
 if TYPE_CHECKING:
-    from collections.abc import Iterator, Sequence
+    from collections.abc import Iterable, Iterator
     from logging import Logger
 
     from beets.library import Item
 
-if sys.version_info >= (3, 10):
-    from typing import TypeAlias
-else:
-    from typing_extensions import TypeAlias
-
 
 MAX_FILENAME_LENGTH = 200
 WINDOWS_MAGIC_PREFIX = "\\\\?\\"
 T = TypeVar("T")
-BytesOrStr = Union[str, bytes]
-PathLike = Union[BytesOrStr, Path]
-Replacements: TypeAlias = "Sequence[tuple[Pattern[str], str]]"
+PathLike = Union[str, bytes, Path]
+Replacements = Sequence[tuple[Pattern[str], str]]
 
 # Here for now to allow for a easy replace later on
 # once we can move to a PathLike (mainly used in importer)
@@ -860,7 +854,9 @@ class CommandOutput(NamedTuple):
     stderr: bytes
 
 
-def command_output(cmd: list[BytesOrStr], shell: bool = False) -> CommandOutput:
+def command_output(
+    cmd: list[str] | list[bytes], shell: bool = False
+) -> CommandOutput:
     """Runs the command and returns its output after it has exited.
 
     Returns a CommandOutput. The attributes ``stdout`` and ``stderr`` contain
@@ -878,8 +874,6 @@ def command_output(cmd: list[BytesOrStr], shell: bool = False) -> CommandOutput:
     This replaces `subprocess.check_output` which can have problems if lots of
     output is sent to stderr.
     """
-    converted_cmd = [os.fsdecode(a) for a in cmd]
-
     devnull = subprocess.DEVNULL
 
     proc = subprocess.Popen(
@@ -894,7 +888,7 @@ def command_output(cmd: list[BytesOrStr], shell: bool = False) -> CommandOutput:
     if proc.returncode:
         raise subprocess.CalledProcessError(
             returncode=proc.returncode,
-            cmd=" ".join(converted_cmd),
+            cmd=" ".join(map(os.fsdecode, cmd)),
             output=stdout + stderr,
         )
     return CommandOutput(stdout, stderr)

beets/util/artresizer.py

@@ -214,9 +214,9 @@ def version(cls) -> tuple[int, int, int]:
         else:
             return cls._version
 
-    convert_cmd: list[str | bytes]
-    identify_cmd: list[str | bytes]
-    compare_cmd: list[str | bytes]
+    convert_cmd: list[str]
+    identify_cmd: list[str]
+    compare_cmd: list[str]
 
     def __init__(self) -> None:
         """Initialize a wrapper around ImageMagick for local image operations.
@@ -265,7 +265,7 @@ def resize(
         # with regards to the height.
         # ImageMagick already seems to default to no interlace, but we include
         # it here for the sake of explicitness.
-        cmd: list[str | bytes] = self.convert_cmd + [
+        cmd: list[str] = self.convert_cmd + [
             syspath(path_in, prefix=False),
             "-resize",
             f"{maxwidth}x>",
@@ -295,7 +295,7 @@ def resize(
         return path_out
 
     def get_size(self, path_in: bytes) -> tuple[int, int] | None:
-        cmd: list[str | bytes] = self.identify_cmd + [
+        cmd: list[str] = self.identify_cmd + [
             "-format",
             "%w %h",
             syspath(path_in, prefix=False),
@@ -480,10 +480,11 @@ def can_write_metadata(self) -> bool:
         return True
 
     def write_metadata(self, file: bytes, metadata: Mapping[str, str]) -> None:
-        assignments = list(
-            chain.from_iterable(("-set", k, v) for k, v in metadata.items())
+        assignments = chain.from_iterable(
+            ("-set", k, v) for k, v in metadata.items()
         )
-        command = self.convert_cmd + [file, *assignments, file]
+        str_file = os.fsdecode(file)
+        command = self.convert_cmd + [str_file, *assignments, str_file]
 
         util.command_output(command)
 

beetsplug/absubmit.py

@@ -137,7 +137,7 @@ def command(self, lib, opts, args):
             )
         else:
             # Get items from arguments
-            items = lib.items(ui.decargs(args))
+            items = lib.items(args)
             self.opts = opts
             util.par_map(self.analyze_submit, items)
 

beetsplug/acousticbrainz.py

@@ -116,7 +116,7 @@ def commands(self):
         )
 
         def func(lib, opts, args):
-            items = lib.items(ui.decargs(args))
+            items = lib.items(args)
             self._fetch_info(
                 items,
                 ui.should_write(),

beetsplug/badfiles.py

@@ -204,7 +204,7 @@ def on_import_task_before_choice(self, task, session):
 
     def command(self, lib, opts, args):
         # Get items from arguments
-        items = lib.items(ui.decargs(args))
+        items = lib.items(args)
         self.verbose = opts.verbose
 
         def check_and_print(item):