refactor util + other misc improvements

Author: amogus07

beets/__init__.py

@@ -13,17 +13,16 @@
 # included in all copies or substantial portions of the Software.
 
 
-from sys import stderr
-
-import confuse
-
+from .config import IncludeLazyConfig, config
 from .util import deprecate_imports
 
-__version__ = "2.5.1"
-__author__ = "Adrian Sampson <[email protected]>"
+__version__: str = "2.5.1"
+__author__: str = "Adrian Sampson <[email protected]>"
+
+__all__: list[str] = ["IncludeLazyConfig", "config"]
 
 
-def __getattr__(name: str):
+def __getattr__(name: str) -> str:
     """Handle deprecated imports."""
     return deprecate_imports(
         old_module=__name__,
@@ -34,23 +33,3 @@ def __getattr__(name: str):
         name=name,
         version="3.0.0",
     )
-
-
-class IncludeLazyConfig(confuse.LazyConfig):
-    """A version of Confuse's LazyConfig that also merges in data from
-    YAML files specified in an `include` setting.
-    """
-
-    def read(self, user=True, defaults=True):
-        super().read(user, defaults)
-
-        try:
-            for view in self["include"]:
-                self.set_file(view.as_filename())
-        except confuse.NotFoundError:
-            pass
-        except confuse.ConfigReadError as err:
-            stderr.write(f"configuration `import` failed: {err.reason}")
-
-
-config = IncludeLazyConfig("beets", __name__)

beets/autotag/distance.py

@@ -9,7 +9,8 @@
 from unidecode import unidecode
 
 from beets import config, metadata_plugins
-from beets.util import as_string, cached_classproperty, get_most_common_tags
+from beets.util import as_string, cached_classproperty
+from beets.util.misc import get_most_common_tags
 
 if TYPE_CHECKING:
     from collections.abc import Iterator, Sequence
@@ -345,7 +346,7 @@ def add_string(self, key: str, str1: str | None, str2: str | None):
         dist = string_dist(str1, str2)
         self.add(key, dist)
 
-    def add_data_source(self, before: str | None, after: str | None) -> None:
+    def add_data_source(self, before: object, after: str | None) -> None:
         if before != after and (
             before or len(metadata_plugins.find_metadata_source_plugins()) > 1
         ):
@@ -384,11 +385,19 @@ def track_distance(
     cached because this function is called many times during the matching
     process and their access comes with a performance overhead.
     """
-    dist = Distance()
+    dist: Distance = Distance()
 
     # Length.
+    info_length: float | None
     if info_length := track_info.length:
-        diff = abs(item.length - info_length) - get_track_length_grace()
+        diff: float = (
+            abs(
+                (item.length - info_length)
+                if isinstance(item.length, (int, float))
+                else 0
+            )
+            - get_track_length_grace()
+        )
         dist.add_ratio("track_length", diff, get_track_length_max())
 
     # Title.

beets/autotag/match.py

@@ -26,7 +26,7 @@
 
 from beets import config, logging, metadata_plugins
 from beets.autotag import AlbumInfo, AlbumMatch, TrackInfo, TrackMatch, hooks
-from beets.util import get_most_common_tags
+from beets.util.misc import get_most_common_tags
 
 from .distance import VA_ARTISTS, distance, track_distance
 

beets/config.py

@@ -0,0 +1,26 @@
+from sys import stderr
+
+import confuse
+from typing_extensions import override
+
+
+class IncludeLazyConfig(confuse.LazyConfig):
+    """A version of Confuse's LazyConfig that also merges in data from
+    YAML files specified in an `include` setting.
+    """
+
+    @override
+    def read(self, user: bool = True, defaults: bool = True) -> None:
+        super().read(user, defaults)
+
+        try:
+            view: confuse.Subview
+            for view in self["include"]:
+                self.set_file(view.as_filename())
+        except confuse.NotFoundError:
+            pass
+        except confuse.ConfigReadError as err:
+            _ = stderr.write(f"configuration `import` failed: {err.reason}")
+
+
+config: IncludeLazyConfig = IncludeLazyConfig("beets", __name__)

beets/logging.py

@@ -37,7 +37,7 @@
     RootLogger,
     StreamHandler,
 )
-from typing import TYPE_CHECKING, Any, Mapping, TypeVar, Union, overload
+from typing import TYPE_CHECKING, Any, TypeVar, Union, overload
 
 __all__ = [
     "DEBUG",
@@ -54,9 +54,11 @@
 ]
 
 if TYPE_CHECKING:
-    T = TypeVar("T")
+    from collections.abc import Mapping
     from types import TracebackType
 
+    T = TypeVar("T")
+
     # see https://github.com/python/typeshed/blob/main/stdlib/logging/__init__.pyi
     _SysExcInfoType = Union[
         tuple[type[BaseException], BaseException, Union[TracebackType, None]],
@@ -144,24 +146,24 @@ def _log(
 class ThreadLocalLevelLogger(Logger):
     """A version of `Logger` whose level is thread-local instead of shared."""
 
-    def __init__(self, name, level=NOTSET):
-        self._thread_level = threading.local()
-        self.default_level = NOTSET
+    def __init__(self, name: str, level: int = NOTSET) -> None:
+        self._thread_level: threading.local = threading.local()
+        self.default_level: int = NOTSET
         super().__init__(name, level)
 
     @property
-    def level(self):
+    def level(self) -> int:
         try:
             return self._thread_level.level
         except AttributeError:
             self._thread_level.level = self.default_level
             return self.level
 
     @level.setter
-    def level(self, value):
+    def level(self, value: int) -> None:
         self._thread_level.level = value
 
-    def set_global_level(self, level):
+    def set_global_level(self, level: int) -> None:
         """Set the level on the current thread + the default value for all
         threads.
         """

beets/ui/__init__.py

@@ -29,10 +29,10 @@
 
 import confuse
 
-from beets import IncludeLazyConfig, config, library, logging, plugins, util
+from beets import library, logging, plugins, util
+from beets.config import config
 from beets.dbcore import db
 from beets.dbcore import query as db_query
-from beets.ui import commands, core
 from beets.ui._common import UserError
 from beets.ui.colors import color_split, colorize, uncolorize
 from beets.ui.core import (
@@ -52,11 +52,12 @@
     should_write,
     show_model_changes,
     split_into_lines,
-    uncolorize,
 )
 
 if TYPE_CHECKING:
-    from beets.library.library import Library
+    from collections.abc import Callable
+
+    from beets.config import IncludeLazyConfig
 
 
 # On Windows platforms, use colorama to support "ANSI" terminal colors.
@@ -77,8 +78,6 @@
     "UserError",
     "color_split",
     "colorize",
-    "commands",
-    "core",
     "input_",
     "input_options",
     "input_select_objects",
@@ -130,7 +129,7 @@ def _configure(options: optparse.Values) -> IncludeLazyConfig:
     # special handling lets specified plugins get loaded before we
     # finish parsing the command line.
     if getattr(options, "config", None) is not None:
-        overlay_path = options.config
+        overlay_path: str | None = options.config
         del options.config
         config.set_file(overlay_path)
     else:
@@ -148,7 +147,7 @@ def _configure(options: optparse.Values) -> IncludeLazyConfig:
             "overlaying configuration: {}", util.displayable_path(overlay_path)
         )
 
-    config_path = config.user_config_path()
+    config_path: str = config.user_config_path()
     if os.path.isfile(config_path):
         log.debug("user configuration: {}", util.displayable_path(config_path))
     else:
@@ -161,10 +160,12 @@ def _configure(options: optparse.Values) -> IncludeLazyConfig:
     return config
 
 
-def _ensure_db_directory_exists(path):
+def _ensure_db_directory_exists(
+    path: util.PathLike,
+):
     if path == b":memory:":  # in memory db
         return
-    newpath = os.path.dirname(path)
+    newpath: bytes | str = os.path.dirname(path)
     if not os.path.isdir(newpath):
         if input_yn(
             f"The database directory {util.displayable_path(newpath)} does not"
@@ -175,16 +176,17 @@ def _ensure_db_directory_exists(path):
 
 def _open_library(config: confuse.LazyConfig) -> library.Library:
     """Create a new library instance from the configuration."""
-    dbpath = util.bytestring_path(config["library"].as_filename())
+    dbpath: bytes = util.bytestring_path(config["library"].as_filename())
     _ensure_db_directory_exists(dbpath)
+    lib: library.Library
     try:
         lib = library.Library(
             dbpath,
             config["directory"].as_filename(),
             get_path_formats(),
             get_replacements(),
         )
-        lib.get_item(0)  # Test database connection.
+        _ = lib.get_item(0)  # Test database connection.
     except (sqlite3.OperationalError, sqlite3.DatabaseError) as db_error:
         log.debug("{}", traceback.format_exc())
         raise UserError(
@@ -199,11 +201,11 @@ def _open_library(config: confuse.LazyConfig) -> library.Library:
     return lib
 
 
-def _raw_main(args: list[str], lib: Library | None = None) -> None:
+def _raw_main(args: list[str], lib: library.Library | None = None) -> None:
     """A helper function for `main` without top-level exception
     handling.
     """
-    parser = SubcommandsOptionParser()
+    parser: SubcommandsOptionParser = SubcommandsOptionParser()
     parser.add_format_option(flags=("--format-item",), target=library.Item)
     parser.add_format_option(flags=("--format-album",), target=library.Album)
     _ = parser.add_option(
@@ -227,7 +229,10 @@ def _raw_main(args: list[str], lib: Library | None = None) -> None:
     )
 
     def parse_csl_callback(
-        option: optparse.Option, _, value: str, parser: SubcommandsOptionParser
+        option: optparse.Option,
+        _: Callable[..., object],
+        value: str,
+        parser: SubcommandsOptionParser,
     ) -> None:
         """Parse a comma-separated list of values."""
         setattr(
@@ -236,36 +241,38 @@ def parse_csl_callback(
             list(filter(None, value.split(","))),
         )
 
-    parser.add_option(
+    _ = parser.add_option(
         "-p",
         "--plugins",
         dest="plugins",
         action="callback",
         callback=parse_csl_callback,
         help="a comma-separated list of plugins to load",
     )
-    parser.add_option(
+    _ = parser.add_option(
         "-P",
         "--disable-plugins",
         dest="disabled_plugins",
         action="callback",
         callback=parse_csl_callback,
         help="a comma-separated list of plugins to disable",
     )
-    parser.add_option(
+    _ = parser.add_option(
         "-h",
         "--help",
         dest="help",
         action="store_true",
         help="show this help message and exit",
     )
-    parser.add_option(
+    _ = parser.add_option(
         "--version",
         dest="version",
         action="store_true",
         help=optparse.SUPPRESS_HELP,
     )
 
+    options: optparse.Values
+    subargs: list[str]
     options, subargs = parser.parse_global_options(args)
 
     # Special case for the `config --edit` command: bypass _setup so
@@ -280,14 +287,17 @@ def parse_csl_callback(
 
         return config_edit()
 
-    test_lib = bool(lib)
+    test_lib: bool = bool(lib)
+    subcommands: list[Subcommand]
     subcommands, lib = _setup(options, lib)
     parser.add_subcommand(*subcommands)
 
+    subcommand: Subcommand
+    suboptions: optparse.Values
     subcommand, suboptions, subargs = parser.parse_subcommand(subargs)
     subcommand.func(lib, suboptions, subargs)
 
-    plugins.send("cli_exit", lib=lib)
+    _ = plugins.send("cli_exit", lib=lib)
     if not test_lib:
         # Clean up the library unless it came from the test harness.
         lib._close()
@@ -308,7 +318,7 @@ def main(args: list[str] | None = None) -> None:
     try:
         _raw_main(args or [])
     except UserError as exc:
-        message = exc.args[0] if exc.args else None
+        message: object = exc.args[0] if exc.args else None
         log.error("error: {}", message)
         sys.exit(1)
     except util.HumanReadableError as exc:
@@ -329,7 +339,7 @@ def main(args: list[str] | None = None) -> None:
     except OSError as exc:
         if exc.errno == errno.EPIPE:
             # "Broken pipe". End silently.
-            sys.stderr.close()
+            _ = sys.stderr.close()
         else:
             raise
     except KeyboardInterrupt: