typing: fix annotations in metadata source plugins

beetsplug/_utils/musicbrainz.py

@@ -454,6 +454,10 @@ class Recording(TypedDict):
     work_relations: NotRequired[list[WorkRelation]]
 
 
+class RecordingWithReleases(Recording):
+    releases: list[BaseRelease]
+
+
 class Track(TypedDict):
     artist_credit: list[ArtistCredit]
     id: str
@@ -501,30 +505,33 @@ class ReleaseRelation(RelationBase):
     release: ReleaseRelationRelease
 
 
-class Release(TypedDict):
-    aliases: list[Alias]
+class BaseRelease(TypedDict):
     artist_credit: list[ArtistCredit]
-    asin: str | None
     barcode: str | None
-    cover_art_archive: CoverArtArchive
     disambiguation: str
-    genres: list[Genre]
     id: str
-    label_info: list[LabelInfo]
-    media: list[Medium]
     packaging: ReleasePackaging | None
     packaging_id: str | None
-    quality: ReleaseQuality
-    release_group: ReleaseGroup
     status: ReleaseStatus | None
     status_id: str | None
-    tags: list[Tag]
+    quality: ReleaseQuality
     text_representation: TextRepresentation
-    title: str
-    artist_relations: NotRequired[list[ArtistRelation]]
     country: NotRequired[str | None]
     date: NotRequired[str]
+    title: str
     release_events: NotRequired[list[ReleaseEvent]]
+
+
+class Release(BaseRelease):
+    aliases: list[Alias]
+    asin: str | None
+    cover_art_archive: CoverArtArchive
+    genres: list[Genre]
+    label_info: list[LabelInfo]
+    media: list[Medium]
+    release_group: ReleaseGroup
+    tags: list[Tag]
+    artist_relations: NotRequired[list[ArtistRelation]]
     release_relations: NotRequired[list[ReleaseRelation]]
     url_relations: NotRequired[list[UrlRelation]]
 
@@ -696,6 +703,14 @@ class MusicBrainzAPI(RequestHandler):
         kwargs.setdefault("includes", RECORDING_INCLUDES)
         return self._lookup("recording", id_, **kwargs)
 
+    def get_base_recording_with_releases(
+        self, id_: str, **kwargs: Unpack[LookupKwargs]
+    ) -> RecordingWithReleases:
+        """Retrieve a recording by its MusicBrainz ID."""
+        kwargs.setdefault("includes", [])
+        kwargs["includes"].append("releases")
+        return self._lookup("recording", id_, **kwargs)
+
     def get_work(self, id_: str, **kwargs: Unpack[LookupKwargs]) -> Work:
         """Retrieve a work by its MusicBrainz ID."""
         return self._lookup("work", id_, **kwargs)

beetsplug/acousticbrainz.py

@@ -125,7 +125,7 @@ class AcousticPlugin(plugins.BeetsPlugin):
             self._fetch_info(
                 items,
                 ui.should_write(),
-                opts.force_refetch or self.config["force"],
+                opts.force_refetch or self.config["force"].get(bool),
             )
 
         cmd.func = func
@@ -262,7 +262,7 @@ class AcousticPlugin(plugins.BeetsPlugin):
         # `composites = {'initial_key': ['B', 'minor']}`.
 
         # The recursive traversal.
-        composites = defaultdict(list)
+        composites = defaultdict[str, list[str]](list)
         yield from self._data_to_scheme_child(data, scheme, composites)
 
         # When composites has been populated, yield the composite attributes

beetsplug/beatport.py

@@ -493,12 +493,12 @@ class BeatportPlugin(MetadataSourcePlugin):
         )
 
     def _get_artist(
-        self, artists: list[tuple[str, str]] | None
+        self, artists: Iterable[tuple[str, str]] | None
     ) -> tuple[str, str | None]:
         """Returns an artist string (all artists) and an artist_id (the main
         artist) for a list of Beatport release or track artists.
         """
-        return self.get_artist(artists=artists, id_key=0, name_key=1)
+        return self.get_artist(artists or [], id_key=0, name_key=1)  # type: ignore[arg-type]
 
     def _get_tracks(self, query: str) -> list[TrackInfo]:
         """Returns a list of TrackInfo objects for a Beatport query."""

beetsplug/chroma.py

@@ -148,7 +148,7 @@ def acoustid_match(log: Logger, path: bytes) -> None:
     # 'countries' to then sort preferred countries first.
     country_patterns = config["match"]["preferred"]["countries"].as_str_seq()
     countries = [re.compile(pat, re.I) for pat in country_patterns]
-    original_year = config["match"]["preferred"]["original_year"]
+    original_year = config["match"]["preferred"]["original_year"].as_str()
     releases.sort(
         key=partial(
             releases_key, countries=countries, original_year=original_year
@@ -170,7 +170,7 @@ def _all_releases(items: Sequence[Item]) -> Iterator[str]:
     which releases the items have in common. Generates release IDs.
     """
     # Count the number of "hits" for each release.
-    relcounts = defaultdict(int)
+    relcounts = defaultdict[str, int](int)
     for item in items:
         if item.path not in _matches:
             continue
@@ -236,11 +236,11 @@ class AcoustidPlugin(MetadataSourcePlugin):
         if self.mb is None:
             return []
 
-        albums = []
-        for relid in prefix(_all_releases(items), MAX_RELEASES):
-            album = self.mb.album_for_id(relid)
-            if album:
-                albums.append(album)
+        albums = [
+            a
+            for relid in prefix(_all_releases(items), MAX_RELEASES)
+            if (a := self.mb.album_for_id(relid))
+        ]
 
         self._log.debug("acoustid album candidates: {}", len(albums))
         return albums
@@ -409,7 +409,8 @@ def submit_items(
     log: Logger, userkey: str, items: Sequence[Item], chunksize: int = 64
 ) -> None:
     """Submit fingerprints for the items to the Acoustid server."""
-    data = []  # The running list of dictionaries to submit.
+    # The running list of dictionaries to submit.
+    data: list[JSONDict] = []
 
     def submit_chunk() -> None:
         """Submit the current accumulated fingerprint data."""
@@ -496,10 +497,10 @@ class ScoredItem:
         self.score = score
 
     def __lt__(self, other: object) -> bool:
-        return self.score < other.score
+        return type(self) is type(other) and self.score < other.score
 
     def __gt__(self, other: object) -> bool:
-        return self.score > other.score
+        return type(self) is type(other) and self.score > other.score
 
     def __str__(self) -> str:
         percent = f"{round(self.score * 100, 2)}%".rjust(6)

beetsplug/deezer.py

@@ -264,20 +264,20 @@ class DeezerPlugin(SearchApiMetadataSourcePlugin[IDResponse]):
                 self._log.debug("No deezer_track_id present for: {}", item)
                 continue
             try:
-                rank = self.fetch_data(
-                    f"{self.track_url}{deezer_track_id}"
-                ).get("rank")
-                self._log.debug(
-                    "Deezer track: {} has {} rank", deezer_track_id, rank
-                )
+                track = self.fetch_data(f"{self.track_url}{deezer_track_id}")
             except Exception as e:
                 self._log.debug("Invalid Deezer track_id: {}", e)
                 continue
-            item.deezer_track_rank = int(rank)
-            item.store()
-            item.deezer_updated = time.time()
-            if write:
-                item.try_write()
+            else:
+                if track and (rank := track.get("rank") is not None):
+                    self._log.debug(
+                        "Deezer track: {} has {} rank", deezer_track_id, rank
+                    )
+                    item.deezer_track_rank = int(rank)
+                    item.store()
+                    item.deezer_updated = time.time()
+                    if write:
+                        item.try_write()
 
     def fetch_data(self, url: str) -> JSONDict | None:
         try:

beetsplug/listenbrainz.py

@@ -21,6 +21,7 @@ from ._utils.playcount import update_play_counts
 from ._utils.requests import TimeoutAndRetrySession
 
 if TYPE_CHECKING:
+    from collections.abc import Iterable
     from pathlib import Path
 
     from beets.library import Library
@@ -146,7 +147,7 @@ class ListenBrainzPlugin(MusicBrainzAPIMixin, BeetsPlugin):
         self._log.info("{} play-counts imported", found)
 
     @staticmethod
-    def _aggregate_listens(tracks: list[Track]) -> list[Track]:
+    def _aggregate_listens(tracks: Iterable[Track]) -> list[Track]:
         """Aggregate individual listen events into per-track play counts.
 
         ListenBrainz returns individual listen events (each with playcount=1).
@@ -304,7 +305,7 @@ class ListenBrainzPlugin(MusicBrainzAPIMixin, BeetsPlugin):
 
         return all_listens
 
-    def get_tracks_from_listens(self, listens: list[Listen]) -> list[Track]:
+    def get_tracks_from_listens(self, listens: Iterable[Listen]) -> list[Track]:
         """Returns a list of tracks from a list of listens."""
         tracks: list[Track] = []
         for track in listens:
@@ -346,7 +347,9 @@ class ListenBrainzPlugin(MusicBrainzAPIMixin, BeetsPlugin):
 
     def get_listenbrainz_playlists(self) -> list[JSONDict]:
         resp = self.get_playlists_createdfor(self.username)
-        playlists = resp.get("playlists")
+        if not resp:
+            return []
+        playlists = resp.get("playlists", [])
         listenbrainz_playlists = []
 
         for playlist in playlists:
@@ -387,7 +390,7 @@ class ListenBrainzPlugin(MusicBrainzAPIMixin, BeetsPlugin):
     def get_tracks_from_playlist(self, playlist: JSONDict) -> list[JSONDict]:
         """This function returns a list of tracks in the playlist."""
         tracks = []
-        for track in playlist.get("playlist").get("track"):
+        for track in playlist.get("playlist", {}).get("track"):
             identifier = track.get("identifier")
             if isinstance(identifier, list):
                 identifier = identifier[0]
@@ -401,23 +404,19 @@ class ListenBrainzPlugin(MusicBrainzAPIMixin, BeetsPlugin):
             )
         return self.get_track_info(tracks)
 
-    def get_track_info(self, tracks: list[JSONDict]) -> list[JSONDict]:
+    def get_track_info(self, tracks: Iterable[JSONDict]) -> list[JSONDict]:
         track_info = []
         for track in tracks:
-            identifier = track.get("identifier")
-            recording = self.mb_api.get_recording(
-                identifier, includes=["releases", "artist-credits"]
-            )
+            identifier = track["identifier"]
+            recording = self.mb_api.get_base_recording_with_releases(identifier)
             title = recording.get("title")
-            artist_credit = recording.get("artist_credit", [])
-            if artist_credit:
-                artist = artist_credit[0].get("artist", {}).get("name")
+            if artist_credit := next(iter(recording["artist_credit"]), None):
+                artist = artist_credit.get("artist", {}).get("name")
             else:
                 artist = None
-            releases = recording.get("releases", [])
-            if releases:
-                album = releases[0].get("title")
-                date = releases[0].get("date")
+            if release := next(iter(recording["releases"]), None):
+                album = release["title"]
+                date = release.get("date")
                 year = date.split("-")[0] if date else None
             else:
                 album = None
@@ -455,5 +454,8 @@ class ListenBrainzPlugin(MusicBrainzAPIMixin, BeetsPlugin):
             f"- {selected_playlist['date']}"
         )
         # Fetch and return tracks from the selected playlist
-        playlist = self.get_playlist(selected_playlist.get("identifier"))
-        return self.get_tracks_from_playlist(playlist)
+        if (identifier := selected_playlist.get("identifier")) and (
+            playlist := self.get_playlist(identifier)
+        ):
+            return self.get_tracks_from_playlist(playlist)
+        return []

beetsplug/mbsubmit.py

@@ -10,13 +10,13 @@ implemented by MusicBrainz yet.
 from __future__ import annotations
 
 import subprocess
+from functools import cached_property
 from typing import TYPE_CHECKING
 
 from beets import ui
 from beets.autotag import Recommendation
 from beets.plugins import BeetsPlugin
 from beets.util import PromptChoice, displayable_path
-from beetsplug.info import print_data
 
 if TYPE_CHECKING:
     import optparse
@@ -73,9 +73,13 @@ class MBSubmitPlugin(BeetsPlugin):
         except OSError as exc:
             self._log.error("Could not open picard, got error:\n{}", exc)
 
+    @cached_property
+    def fmt(self) -> str:
+        return self.config["format"].as_str()
+
     def print_tracks(self, session: ImportSession, task: ImportTask) -> None:
         for i in sorted(task.items, key=lambda i: i.track):
-            print_data(None, i, self.config["format"].as_str())
+            ui.print_(format(i, self.fmt))
 
     def commands(self) -> list[ui.Subcommand]:
         """Add beet UI commands for mbsubmit."""
@@ -94,4 +98,4 @@ class MBSubmitPlugin(BeetsPlugin):
     def _mbsubmit(self, items: Sequence[Item]) -> None:
         """Print track information to be submitted to MusicBrainz."""
         for i in sorted(items, key=lambda i: i.track):
-            print_data(None, i, self.config["format"].as_str())
+            ui.print_(format(i, self.fmt))

beetsplug/spotify.py

@@ -27,7 +27,7 @@ from beets.metadata_plugins import IDResponse, SearchApiMetadataSourcePlugin
 from beets.util import chunks
 
 if TYPE_CHECKING:
-    from collections.abc import Mapping, Sequence
+    from collections.abc import Iterable, Mapping, Sequence
 
     from beets.library import Item, Library
     from beets.metadata_plugins import QueryType, SearchParams
@@ -319,7 +319,7 @@ class SpotifyPlugin(
             raise APIError("Request failed.")
 
     def _multi_artist_credit(
-        self, artists: list[dict[str | int, str]]
+        self, artists: Iterable[dict[str | int, str]]
     ) -> tuple[list[str], list[str]]:
         """Given a list of artist dictionaries, accumulate data into a pair
         of lists: the first being the artist names, and the second being the
@@ -607,7 +607,7 @@ class SpotifyPlugin(
         return True
 
     def _match_library_tracks(
-        self, library: Library, keywords: list[str]
+        self, library: Library, keywords: Sequence[str]
     ) -> list[SearchResponseAlbums | SearchResponseTracks] | None:
         """Get simplified track object dicts for library tracks.
 
@@ -883,7 +883,9 @@ class SpotifyPlugin(
             for item, _ in items_to_update:
                 item.store()
 
-    def track_info(self, track_id: str) -> tuple[Any, Any, Any, Any]:
+    def track_info(
+        self, track_id: str
+    ) -> tuple[int | None, str | None, str | None, str | None]:
         """Fetch a track's popularity and external IDs using its Spotify ID."""
         track_data = self._handle_response(
             "get", f"{self.track_url}/{track_id}"

test/plugins/test_listenbrainz.py

@@ -43,7 +43,7 @@ class TestListenBrainzPlugin(ConfigMixin):
 
     def test_get_track_info(self, plugin, requests_mock):
         requests_mock.get(
-            "/ws/2/recording/id1?inc=releases%2Bartist-credits",
+            "/ws/2/recording/id1?inc=releases",
             json={
                 "title": "T",
                 "artist-credit": [],