typing: type the rest of plugins

beets/dbcore/query.py

@@ -50,7 +50,7 @@ class InvalidQueryError(ParsingError):
     def __init__(
         self, query: str | Sequence[str] | Query | None, explanation: Exception
     ) -> None:
-        if isinstance(query, list):
+        if isinstance(query, Sequence) and not isinstance(query, str):
             query = " ".join(query)
         message = f"'{query}': {explanation}"
         super().__init__(message)

beets/library/library.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import re
+from collections.abc import Sequence
 from contextlib import contextmanager
 from functools import cached_property
 from pathlib import Path
@@ -21,7 +22,7 @@
 from .queries import parse_query_parts, parse_query_string
 
 if TYPE_CHECKING:
-    from collections.abc import Iterator, Sequence
+    from collections.abc import Iterator
 
     from beets.dbcore.sort import Sort
     from beets.util import PathLike, Replacements
@@ -103,7 +104,7 @@ def add(self, obj: LibModel) -> int | None:
         self._memotable = {}
         return obj.id
 
-    def add_album(self, items: list[Item]) -> Album:
+    def add_album(self, items: Sequence[Item]) -> Album:
         """Create a new album consisting of a list of items.
 
         The items are added to the database if they don't yet have an
@@ -156,7 +157,7 @@ def _fetch(
                     parsed_query, parsed_sort = parse_query_string(
                         query, model_cls
                     )
-                elif isinstance(query, (list, tuple)):
+                elif isinstance(query, Sequence):
                     parsed_query, parsed_sort = parse_query_parts(
                         query, model_cls
                     )

beets/library/models.py

@@ -1411,7 +1411,8 @@ def tmpl_aunique(
         if memoval is not None:
             return memoval
 
-        album: Album = self.lib.get_album(album_id)  # type: ignore[assignment]
+        if not (album := self.lib.get_album(album_id)):
+            return ""
 
         return self._tmpl_unique(
             "aunique",

beets/ui/commands/import_/session.py

@@ -297,7 +297,7 @@ def _get_choices(self, task: ImportTask) -> list[PromptChoice]:
         return choices + extra_choices
 
 
-def summarize_items(items: list[Item], singleton: bool) -> str:
+def summarize_items(items: Sequence[Item], singleton: bool) -> str:
     """Produces a brief summary line describing a set of items. Used for
     manually resolving duplicates during import.
 

beets/util/__init__.py

@@ -404,19 +404,14 @@ def displayable_path(
     path: PathLike | Iterable[PathLike], separator: str = "; "
 ) -> str:
     """Attempts to decode a bytestring path to a unicode object for the
-    purpose of displaying it to the user. If the `path` argument is a
-    list or a tuple, the elements are joined with `separator`.
+    purpose of displaying it to the user. If the `path` argument is an
+    iterable, the elements are joined with `separator`.
     """
 
-    if isinstance(path, (list, tuple)):
-        return separator.join(displayable_path(p) for p in path)
-    if isinstance(path, str):
-        return path
-    if not isinstance(path, bytes):
-        # A non-string object: just get its unicode representation.
-        return str(path)
+    if isinstance(path, (Path, str, bytes)):
+        return os.fsdecode(path)
 
-    return os.fsdecode(path)
+    return separator.join(displayable_path(p) for p in path)
 
 
 def syspath(path: PathLike) -> str:
@@ -850,7 +845,7 @@ class CommandOutput(NamedTuple):
 
 
 def command_output(
-    cmd: list[str] | list[bytes], shell: bool = False
+    cmd: Sequence[str] | Sequence[bytes], shell: bool = False
 ) -> CommandOutput:
     """Runs the command and returns its output after it has exited.
 
@@ -943,7 +938,9 @@ def editor_command() -> str:
     )
 
 
-def interactive_open(targets: Sequence[str], command: str) -> None:
+def interactive_open(
+    targets: Sequence[Path | str | bytes], command: str
+) -> None:
     """Open the files in `targets` by `exec`ing a new `command`, given
     as a Unicode string. (The new program takes over, and Python
     execution ends: this does not fork a subprocess.)
@@ -958,11 +955,10 @@ def interactive_open(targets: Sequence[str], command: str) -> None:
     except ValueError:  # Malformed shell tokens.
         args = [command]
 
-    args.insert(0, args[0])  # for argv[0]
-
-    args += targets
+    first, *rest = args
 
-    os.execlp(*args)
+    # 'first' is duplicated because of argv[0]
+    os.execlp(*[first, first, *rest, *targets])
 
 
 def case_sensitive(path: AnyStr) -> bool:

beetsplug/_utils/vfs.py

@@ -9,6 +9,8 @@
 from beets import util
 
 if TYPE_CHECKING:
+    from collections.abc import Sequence
+
     from beets.library import Library
 
 
@@ -20,7 +22,7 @@ class Node(NamedTuple):
     # Maps directory names to child nodes.
 
 
-def _insert(node: Node, path: list[str], itemid: int):
+def _insert(node: Node, path: Sequence[str], itemid: int) -> None:
     """Insert an item into a virtual filesystem node."""
     if len(path) == 1:
         # Last component. Insert file.

beetsplug/absubmit.py

@@ -17,7 +17,11 @@
 from beets.exceptions import UserError
 
 if TYPE_CHECKING:
-    from beets.library import Library
+    from collections.abc import Sequence
+
+    from beets.library import Item, Library
+
+    from ._typing import JSONDict
 
 
 class ABSubmitCLIOpts(Protocol):
@@ -33,7 +37,7 @@ class ABSubmitError(Exception):
     """Raised when failing to analyse file with extractor."""
 
 
-def call(args):
+def call(args: Sequence[str]) -> bytes:
     """Execute the command and return its output.
 
     Raise a AnalysisABSubmitError on failure.
@@ -45,7 +49,7 @@ def call(args):
 
 
 class AcousticBrainzSubmitPlugin(plugins.BeetsPlugin):
-    def __init__(self):
+    def __init__(self) -> None:
         super().__init__()
 
         self._log.warning("This plugin is deprecated.")
@@ -54,19 +58,18 @@ def __init__(self):
             {"extractor": "", "force": False, "pretend": False, "base_url": ""}
         )
 
-        self.extractor = self.config["extractor"].as_str()
-        if self.extractor:
-            self.extractor = util.normpath(self.extractor)
+        if extractor := self.config["extractor"].as_str():
+            extractor = os.fsdecode(util.normpath(extractor))
             # Explicit path to extractor
-            if not os.path.isfile(self.extractor):
+            if not os.path.isfile(extractor):
                 raise UserError(
-                    f"Extractor command does not exist: {self.extractor}."
+                    f"Extractor command does not exist: {extractor}."
                 )
         else:
             # Implicit path to extractor, search for it in path
-            self.extractor = "streaming_extractor_music"
+            extractor = "streaming_extractor_music"
             try:
-                call([self.extractor])
+                call([extractor])
             except OSError:
                 raise UserError(
                     "No extractor command found: please install the extractor"
@@ -79,13 +82,21 @@ def __init__(self):
 
             # Get the executable location on the system, which we need
             # to calculate the SHA-1 hash.
-            self.extractor = shutil.which(self.extractor)
+            if extractor_cmd_path := shutil.which(extractor):
+                extractor = extractor_cmd_path
+            else:
+                raise UserError(
+                    f"Path to extractor command {extractor} not found"
+                )
+
+        self.extractor = extractor
 
         # Calculate extractor hash.
-        self.extractor_sha = hashlib.sha1()
-        with open(self.extractor, "rb") as extractor:
-            self.extractor_sha.update(extractor.read())
-        self.extractor_sha = self.extractor_sha.hexdigest()
+        extractor_sha = hashlib.sha1()
+        if extractor:
+            with open(extractor, "rb") as f:
+                extractor_sha.update(f.read())
+        self.extractor_sha = extractor_sha.hexdigest()
 
         self.url = ""
         base_url = self.config["base_url"].as_str()
@@ -99,7 +110,7 @@ def __init__(self):
                 base_url = f"{base_url}/"
             self.url = f"{base_url}{{mbid}}/low-level"
 
-    def commands(self):
+    def commands(self) -> list[ui.Subcommand]:
         cmd = ui.Subcommand(
             "absubmit", help="calculate and submit AcousticBrainz analysis"
         )
@@ -139,12 +150,12 @@ def command(
         self.opts = opts
         util.par_map(self.analyze_submit, items)
 
-    def analyze_submit(self, item):
+    def analyze_submit(self, item: Item) -> None:
         analysis = self._get_analysis(item)
         if analysis:
             self._submit_data(item, analysis)
 
-    def _get_analysis(self, item):
+    def _get_analysis(self, item: Item) -> JSONDict | None:
         mbid = item["mb_trackid"]
 
         # Avoid re-analyzing files that already have AB data.
@@ -175,13 +186,11 @@ def _get_analysis(self, item):
                 call([self.extractor, util.syspath(item.path), filename])
             except ABSubmitError as e:
                 self._log.warning(
-                    "Failed to analyse {item} for AcousticBrainz: {error}",
-                    item=item,
-                    error=e,
+                    "Failed to analyse {} for AcousticBrainz: {}", item, e
                 )
                 return None
-            with open(filename) as tmp_file:
-                analysis = json.load(tmp_file)
+            with open(filename) as f:
+                analysis = json.load(f)
             # Add the hash to the output.
             analysis["metadata"]["version"]["essentia_build_sha"] = (
                 self.extractor_sha
@@ -195,7 +204,7 @@ def _get_analysis(self, item):
                 if e.errno != errno.ENOENT:
                     raise
 
-    def _submit_data(self, item, data):
+    def _submit_data(self, item: Item, data: JSONDict) -> None:
         mbid = item["mb_trackid"]
         headers = {"Content-Type": "application/json"}
         response = requests.post(
@@ -208,10 +217,9 @@ def _submit_data(self, item, data):
             except (ValueError, KeyError) as e:
                 message = f"unable to get error message: {e}"
             self._log.error(
-                "Failed to submit AcousticBrainz analysis of {item}: "
-                "{message}).",
-                item=item,
-                message=message,
+                "Failed to submit AcousticBrainz analysis for {}: {}.",
+                item,
+                message,
             )
         else:
             self._log.debug(

beetsplug/advancedrewrite.py

@@ -18,15 +18,21 @@
 from .rewrite import apply_rewrite_rules
 
 if TYPE_CHECKING:
-    from collections.abc import Iterator
+    from collections.abc import Callable, Iterator
+
+    from beets.library import LibModel
 
 
 class AdvancedRewriteConfig(TypedDict):
     match: str
     replacements: dict[str, str | list[str]]
 
 
-def rewriter(field, simple_rules, advanced_rules):
+def rewriter(
+    field: str,
+    simple_rules: list[tuple[re.Pattern[str], str]],
+    advanced_rules: list[tuple[AndQuery, str | list[str]]],
+) -> Callable[[LibModel], str]:
     """Template field function factory.
 
     Create a template field function that rewrites the given field
@@ -35,7 +41,7 @@ def rewriter(field, simple_rules, advanced_rules):
     ``advanced_rules`` must be a list of (query, replacement) pairs.
     """
 
-    def fieldfunc(item):
+    def fieldfunc(item: LibModel) -> str:
         value = item._values_fixed[field]
         if (new_value := apply_rewrite_rules(value, simple_rules)) != value:
             # Rewrite activated.
@@ -44,7 +50,9 @@ def fieldfunc(item):
         for query, replacement in advanced_rules:
             if query.match(item):
                 # Rewrite activated.
-                return replacement
+                # TODO: BeetsPlugin.template_fields and album_template_fields
+                # require return value 'str' but 'list' here is legit too
+                return replacement  # type: ignore[return-value]
         # Not activated; return original value.
         return value
 

beetsplug/albumtypes.py

@@ -15,7 +15,7 @@
 class AlbumTypesPlugin(BeetsPlugin):
     """Adds an album template field for formatted album types."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         """Init AlbumTypesPlugin."""
         super().__init__()
         self.album_template_fields["atypes"] = self._atypes
@@ -34,7 +34,7 @@ def __init__(self):
             }
         )
 
-    def _atypes(self, item: Album):
+    def _atypes(self, item: Album) -> str:
         """Returns a formatted string based on album's types."""
         types = self.config["types"].as_pairs()
         ignore_va = self.config["ignore_va"].as_str_seq()

beetsplug/autobpm.py

@@ -12,6 +12,8 @@
 from beets.util.deprecation import deprecate_for_user
 
 if TYPE_CHECKING:
+    from collections.abc import Iterable
+
     from beets.importer import ImportTask
     from beets.library import Item, Library
 
@@ -88,7 +90,7 @@ def imported(self, _, task: ImportTask) -> None:
 
     def calculate_bpm(
         self,
-        items: list[Item],
+        items: Iterable[Item],
         write: bool = False,
         force: bool = False,
         quiet: bool = False,