typing: type beets core

beets/__init__.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from sys import stderr
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 import confuse
 
@@ -14,7 +14,7 @@
 __author__ = "Adrian Sampson <adrian@radbox.org>"
 
 
-def __getattr__(name: str):
+def __getattr__(name: str) -> Any:
     """Handle deprecated imports."""
     return deprecate_imports(
         __name__, {"art": "beetsplug._utils", "vfs": "beetsplug._utils"}, name

beets/autotag/__init__.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from importlib import import_module
+from typing import Any
 
 # Parts of external interface.
 from beets.util.deprecation import deprecate_for_maintainers, deprecate_imports
@@ -22,7 +23,7 @@
 from .source import Source
 
 
-def __getattr__(name: str):
+def __getattr__(name: str) -> Any:
     if name == "current_metadata":
         deprecate_for_maintainers(
             f"'beets.autotag.{name}'", "'beets.util.get_most_common_tags'"

beets/autotag/distance.py

@@ -208,29 +208,38 @@ def items(self) -> list[tuple[str, float]]:
     def __hash__(self) -> int:
         return id(self)
 
-    def __eq__(self, other) -> bool:
+    def __eq__(self, other: object) -> bool:
         return self.distance == other
 
     # Behave like a float.
 
-    def __lt__(self, other) -> bool:
-        return self.distance < other
+    def __lt__(self, other: object) -> bool:
+        if isinstance(other, (int, float, Distance)):
+            return self.distance < other
+
+        return NotImplemented
 
     def __float__(self) -> float:
         return self.distance
 
-    def __sub__(self, other) -> float:
-        return self.distance - other
+    def __sub__(self, other: object) -> float:
+        if isinstance(other, (int, float, Distance)):
+            return self.distance - other
+
+        return NotImplemented
+
+    def __rsub__(self, other: object) -> float:
+        if isinstance(other, (int, float, Distance)):
+            return other - self.distance
 
-    def __rsub__(self, other) -> float:
-        return other - self.distance
+        return NotImplemented
 
     def __str__(self) -> str:
         return f"{self.distance:.2f}"
 
     # Behave like a dict.
 
-    def __getitem__(self, key) -> float:
+    def __getitem__(self, key: str) -> float:
         """Returns the weighted distance for a named penalty."""
         dist = sum(self._penalties[key]) * self._weights[key]
         dist_max = self.max_distance
@@ -247,7 +256,7 @@ def __len__(self) -> int:
     def keys(self) -> KeysView[str]:
         return dict.fromkeys(key for key, _ in self.items()).keys()
 
-    def update(self, dist: Distance):
+    def update(self, dist: Distance) -> None:
         """Adds all the distance penalties from `dist`."""
         if not isinstance(dist, Distance):
             raise ValueError(
@@ -267,7 +276,7 @@ def _eq(self, value1: re.Pattern[str] | Any, value2: Any) -> bool:
             return bool(value1.match(value2))
         return value1 == value2
 
-    def add(self, key: str, dist: float):
+    def add(self, key: str, dist: float) -> None:
         """Adds a distance penalty. `key` must correspond with a
         configured weight setting. `dist` must be a float between 0.0
         and 1.0, and will be added to any existing distance penalties
@@ -279,7 +288,7 @@ def add(self, key: str, dist: float):
 
     def add_equality(
         self, key: str, value: Any, options: list[Any] | tuple[Any, ...] | Any
-    ):
+    ) -> None:
         """Adds a distance penalty of 1.0 if `value` doesn't match any
         of the values in `options`. If an option is a compiled regular
         expression, it will be considered equal if it matches against
@@ -295,7 +304,7 @@ def add_equality(
             dist = 1.0
         self.add(key, dist)
 
-    def add_expr(self, key: str, expr: bool):
+    def add_expr(self, key: str, expr: bool) -> None:
         """Adds a distance penalty of 1.0 if `expr` evaluates to True,
         or 0.0.
         """
@@ -304,7 +313,7 @@ def add_expr(self, key: str, expr: bool):
         else:
             self.add(key, 0.0)
 
-    def add_number(self, key: str, number1: int, number2: int):
+    def add_number(self, key: str, number1: int, number2: int) -> None:
         """Adds a distance penalty of 1.0 for each number of difference
         between `number1` and `number2`, or 0.0 when there is no
         difference. Use this when there is no upper limit on the
@@ -319,7 +328,7 @@ def add_number(self, key: str, number1: int, number2: int):
 
     def add_priority(
         self, key: str, value: Any, options: list[Any] | tuple[Any, ...] | Any
-    ):
+    ) -> None:
         """Adds a distance penalty that corresponds to the position at
         which `value` appears in `options`. A distance penalty of 0.0
         for the first option, or 1.0 if there is no matching option. If
@@ -337,7 +346,9 @@ def add_priority(
             dist = 1.0
         self.add(key, dist)
 
-    def add_ratio(self, key: str, number1: int | float, number2: int | float):
+    def add_ratio(
+        self, key: str, number1: int | float, number2: int | float
+    ) -> None:
         """Adds a distance penalty for `number1` as a ratio of `number2`.
         `number1` is bound at 0 and `number2`.
         """
@@ -348,7 +359,7 @@ def add_ratio(self, key: str, number1: int | float, number2: int | float):
             dist = 0.0
         self.add(key, dist)
 
-    def add_string(self, key: str, str1: str | None, str2: str | None):
+    def add_string(self, key: str, str1: str | None, str2: str | None) -> None:
         """Adds a distance penalty based on the edit distance between
         `str1` and `str2`.
         """

beets/autotag/match.py

@@ -330,7 +330,7 @@ def _sort_candidates(candidates: Iterable[AnyMatch]) -> Sequence[AnyMatch]:
 
 def _add_candidate(
     source: Source, results: Candidates[AlbumMatch], info: AlbumInfo
-):
+) -> None:
     """Given a candidate AlbumInfo object, attempt to add the candidate
     to the output dictionary of AlbumMatch objects. This involves
     checking the track count, ordering the items, checking for

beets/context.py

@@ -1,5 +1,11 @@
+from __future__ import annotations
+
 from contextlib import contextmanager
 from contextvars import ContextVar
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
 
 # Holds the music dir context
 _music_dir_var: ContextVar[bytes] = ContextVar("music_dir", default=b"")
@@ -16,7 +22,7 @@ def set_music_dir(value: bytes) -> None:
 
 
 @contextmanager
-def music_dir(value: bytes):
+def music_dir(value: bytes) -> Iterator[None]:
     """Temporarily bind the active music directory for query parsing."""
     token = _music_dir_var.set(value)
     try:

beets/dbcore/db.py

@@ -653,7 +653,7 @@ def remove(self) -> None:
                 f"DELETE FROM {self._flex_table} WHERE entity_id=?", (self.id,)
             )
 
-    def add(self, db: D | None = None):
+    def add(self, db: D | None = None) -> None:
         """Add the object to the library database. This object must be
         associated with a database; you can provide one via the `db`
         parameter or use the currently associated database.
@@ -681,7 +681,7 @@ def add(self, db: D | None = None):
 
     def formatted(
         self,
-        included_keys: str = FormattedMapping.ALL_KEYS,
+        included_keys: str | list[str] = FormattedMapping.ALL_KEYS,
         for_path: bool = False,
     ) -> FormattedMapping:
         """Get a mapping containing all values on this object formatted

beets/library/__init__.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from beets.util.deprecation import deprecate_imports
 
 from .exceptions import FileOperationError, ReadError, WriteError
@@ -12,7 +14,7 @@
 )
 
 
-def __getattr__(name: str):
+def __getattr__(name: str) -> Any:
     return deprecate_imports(__name__, NEW_MODULE_BY_NAME, name)
 
 

beets/library/models.py

@@ -91,10 +91,13 @@ def store(self, fields: Iterable[str] | None = None) -> None:
         super().store(fields)
         plugins.send("database_change", lib=self.db, model=self)
 
-    def remove(self) -> None:
+    def _remove(self) -> None:
         super().remove()
         plugins.send("database_change", lib=self.db, model=self)
 
+    def remove(self, delete: bool = False) -> None:
+        raise NotImplementedError
+
     def add(self, lib: Library | None = None) -> None:
         # super().add() calls self.store(), which sends `database_change`,
         # so don't do it here
@@ -386,7 +389,7 @@ def remove(self, delete: bool = False, with_items: bool = True) -> None:
 
         Set with_items to False to avoid removing the album's items.
         """
-        super().remove()
+        super()._remove()
 
         # Send a 'album_removed' signal to plugins
         plugins.send("album_removed", album=self)
@@ -1121,7 +1124,7 @@ def remove(self, delete: bool = False, with_album: bool = True) -> None:
         If `with_album`, then the item's album (if any) is removed
         if the item was the last in the album.
         """
-        super().remove()
+        super()._remove()
 
         # Remove the album if it is empty.
         if with_album:

beets/logging.py

@@ -28,7 +28,7 @@
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
-    from logging import RootLogger
+    from logging import LogRecord, RootLogger
     from types import TracebackType
 
     T = TypeVar("T")
@@ -113,7 +113,7 @@ class LegacyFormatter(Formatter):
         "beets.musicbrainz.sub"     "musicbrainz.sub: msg"
     """
 
-    def format(self, record):
+    def format(self, record: LogRecord) -> str:
         parts = record.name.split(".")
         record.legacy_prefix = (
             f"{'.'.join(parts[1:])}: " if len(parts) > 1 else ""
@@ -134,12 +134,14 @@ class StrFormatLogger(Logger):
     """
 
     class _LogMessage:
-        def __init__(self, msg: str, args: _ArgsType, kwargs: dict[str, Any]):
+        def __init__(
+            self, msg: str, args: _ArgsType, kwargs: dict[str, Any]
+        ) -> None:
             self.msg = msg
             self.args = args
             self.kwargs = kwargs
 
-        def __str__(self):
+        def __str__(self) -> str:
             args = [_logsafe(a) for a in self.args]
             kwargs = {k: _logsafe(v) for (k, v) in self.kwargs.items()}
             return self.msg.format(*args, **kwargs)
@@ -154,7 +156,7 @@ def _log(
         stack_info: bool = False,
         stacklevel: int = 2,
         **kwargs,
-    ):
+    ) -> None:
         """Log msg.format(*args, **kwargs)"""
 
         if isinstance(msg, str):
@@ -174,24 +176,24 @@ def _log(
 class ThreadLocalLevelLogger(Logger):
     """A version of `Logger` whose level is thread-local instead of shared."""
 
-    def __init__(self, name, level=NOTSET):
+    def __init__(self, name: str, level: int = NOTSET) -> None:
         self._thread_level = threading.local()
         self.default_level = 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.
         """
@@ -223,7 +225,7 @@ def extra_debug(self, msg: str, *args: Any, **kwargs: Any) -> None:
 def getLogger(name: str) -> BeetsLogger: ...
 @overload
 def getLogger(name: None = ...) -> RootLogger: ...
-def getLogger(name=None) -> BeetsLogger | RootLogger:  # noqa: N802
+def getLogger(name: str | None = None) -> BeetsLogger | RootLogger:  # noqa: N802
     if name:
         return my_manager.getLogger(name)  # type: ignore[return-value]
     return Logger.root

beets/metadata_plugins.py

@@ -60,7 +60,9 @@ def get_metadata_source(name: str) -> MetadataSourcePlugin | None:
 
 
 @contextmanager
-def maybe_handle_plugin_error(plugin: MetadataSourcePlugin, method_name: str):
+def maybe_handle_plugin_error(
+    plugin: MetadataSourcePlugin, method_name: str
+) -> Iterator[None]:
     """Safely call a plugin method, catching and logging exceptions."""
     if config["raise_on_error"]:
         yield

beets/plugins.py

@@ -73,7 +73,7 @@ class PluginImportError(ImportError):
     from other errors.
     """
 
-    def __init__(self, name: str):
+    def __init__(self, name: str) -> None:
         super().__init__(f"Could not import plugin {name}")
 
 
@@ -162,7 +162,7 @@ def __init_subclass__(cls) -> None:
         ):
             setattr(cls, name, method)
 
-    def __init__(self, name: str | None = None):
+    def __init__(self, name: str | None = None) -> None:
         """Perform one-time plugin setup."""
 
         self.name = name or self.__module__.split(".")[-1]