typing: type metadata source plugins: discogs, deezer, beatport, spotify chroma, listenbrainz, mbsubmit, spotify

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]]
 
@@ -618,7 +625,7 @@ def request(self, *args, **kwargs) -> Response:
         kwargs["params"]["fmt"] = "json"
         return super().request(*args, **kwargs)
 
-    def get_json(self, *args, **kwargs):
+    def get_json(self, *args, **kwargs) -> JSONDict:
         """Fetch JSON data from MusicBrainz and normalize its field names."""
         return self._normalize_data(super().get_json(*args, **kwargs))
 
@@ -696,6 +703,14 @@ def get_recording(
         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/_utils/requests.py

@@ -90,7 +90,7 @@ def __init__(self, *args, **kwargs) -> None:
         self.mount("https://", adapter)
         self.mount("http://", adapter)
 
-    def request(self, *args, **kwargs):
+    def request(self, *args, **kwargs) -> requests.Response:
         """Execute HTTP request with automatic timeout and status validation.
 
         Ensures all requests have a timeout (defaults to 10 seconds) and raises
@@ -115,7 +115,7 @@ class RateLimitAdapter(HTTPAdapter):
     Override `_wait_time()` for custom strategies (token bucket, burst, etc.).
     """
 
-    def __init__(self, rate_limit: float = 0.25, **kwargs):
+    def __init__(self, rate_limit: float = 0.25, **kwargs) -> None:
         super().__init__(**kwargs)
         self.rate_limit = rate_limit
         self._last_request_time = 0.0
@@ -125,7 +125,9 @@ def _wait_time(self, elapsed: float) -> float:
         """Return seconds to wait. Override for custom rate limiting."""
         return max(0, self.rate_limit - elapsed)
 
-    def send(self, request: requests.PreparedRequest, *args, **kwargs):
+    def send(
+        self, request: requests.PreparedRequest, *args, **kwargs
+    ) -> requests.Response:
         with self._lock:
             elapsed = time.monotonic() - self._last_request_time
             wait = self._wait_time(elapsed)
@@ -222,6 +224,6 @@ def delete(self, *args, **kwargs) -> requests.Response:
         """Perform HTTP DELETE request with automatic error handling."""
         return self.request("delete", *args, **kwargs)
 
-    def get_json(self, *args, **kwargs):
+    def get_json(self, *args, **kwargs) -> Any:
         """Fetch and parse JSON data from an HTTP endpoint."""
         return self.get(*args, **kwargs).json()

beetsplug/acousticbrainz.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from collections import defaultdict
-from typing import TYPE_CHECKING, ClassVar, Protocol
+from typing import TYPE_CHECKING, Any, ClassVar, Protocol
 
 import requests
 
@@ -12,8 +12,12 @@
 from beets.exceptions import UserError
 
 if TYPE_CHECKING:
+    from collections.abc import Iterator, Sequence
+
     from beets.importer import ImportSession, ImportTask
-    from beets.library import Library
+    from beets.library import Item, Library
+
+    from ._typing import JSONDict
 
 
 class AcousticBrainzCLIOpts(Protocol):
@@ -101,7 +105,7 @@ def __init__(self) -> None:
         if self.config["auto"]:
             self.register_listener("import_task_files", self.import_task_files)
 
-    def commands(self):
+    def commands(self) -> list[ui.Subcommand]:
         cmd = ui.Subcommand(
             "acousticbrainz", help="fetch metadata from AcousticBrainz"
         )
@@ -121,7 +125,7 @@ def func(
             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
@@ -133,7 +137,7 @@ def import_task_files(
         """Function is called upon beet import."""
         self._fetch_info(task.imported_items(), False, True)
 
-    def _get_data(self, mbid):
+    def _get_data(self, mbid: str) -> JSONDict:
         if not self.base_url:
             raise UserError(
                 "This plugin is deprecated since AcousticBrainz has shut "
@@ -161,7 +165,9 @@ def _get_data(self, mbid):
 
         return data
 
-    def _fetch_info(self, items, write, force):
+    def _fetch_info(
+        self, items: Sequence[Item], write: bool, force: bool
+    ) -> None:
         """Fetch additional information from AcousticBrainz for the `item`s."""
         tags = self.config["tags"].as_str_seq()
         for item in items:
@@ -200,7 +206,9 @@ def _fetch_info(self, items, write, force):
                 if write:
                     item.try_write()
 
-    def _map_data_to_scheme(self, data, scheme):
+    def _map_data_to_scheme(
+        self, data: JSONDict, scheme: JSONDict
+    ) -> Iterator[tuple[str, Any]]:
         """Given `data` as a structure of nested dictionaries, and
         `scheme` as a structure of nested dictionaries , `yield` tuples
         `(attr, val)` where `attr` and `val` are corresponding leaf
@@ -254,15 +262,20 @@ def _map_data_to_scheme(self, data, scheme):
         # `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
         # by joining their parts.
         for composite_attr, value_parts in composites.items():
             yield composite_attr, " ".join(value_parts)
 
-    def _data_to_scheme_child(self, subdata, subscheme, composites):
+    def _data_to_scheme_child(
+        self,
+        subdata: JSONDict,
+        subscheme: JSONDict,
+        composites: dict[str, list[str]],
+    ) -> Iterator[tuple[str, Any]]:
         """The recursive business logic of :meth:`_map_data_to_scheme`:
         Traverse two structures of nested dictionaries in parallel and `yield`
         tuples of corresponding leaf nodes.
@@ -300,7 +313,7 @@ def _data_to_scheme_child(self, subdata, subscheme, composites):
                 )
 
 
-def _generate_urls(base_url, mbid):
+def _generate_urls(base_url: str, mbid: str) -> Iterator[str]:
     """Generates AcousticBrainz end point urls for given `mbid`."""
     for level in LEVELS:
         yield f"{base_url}{mbid}{level}"

beetsplug/beatport.py

@@ -44,7 +44,11 @@ class BeatportClient:
     _api_base = "https://oauth-api.beatport.com"
 
     def __init__(
-        self, c_key, c_secret, auth_key=None, auth_secret=None
+        self,
+        c_key: str,
+        c_secret: str,
+        auth_key: str | None = None,
+        auth_secret: str | None = None,
     ) -> None:
         """Initiate the client with OAuth information.
 
@@ -118,7 +122,7 @@ def search(
         self,
         query: str,
         release_type: Literal["release", "track"],
-        details=True,
+        details: bool = True,
     ) -> Iterator[BeatportRelease | BeatportTrack]:
         """Perform a search of the Beatport catalogue.
 
@@ -389,7 +393,7 @@ def item_candidates(
             self._log.debug("API Error: {} (query: {})", e, query)
             return []
 
-    def album_for_id(self, album_id: str):
+    def album_for_id(self, album_id: str) -> AlbumInfo | None:
         """Fetches a release by its Beatport ID and returns an AlbumInfo object
         or None if the query is not a valid ID or release is not found.
         """
@@ -404,7 +408,7 @@ def album_for_id(self, album_id: str):
             return self._get_album_info(release)
         return None
 
-    def track_for_id(self, track_id: str):
+    def track_for_id(self, track_id: str) -> TrackInfo | None:
         """Fetches a track by its Beatport ID and returns a TrackInfo object
         or None if the track is not a valid Beatport ID or track is not found.
         """
@@ -488,13 +492,15 @@ def _get_track_info(self, track: BeatportTrack) -> TrackInfo:
             genres=track.genres,
         )
 
-    def _get_artist(self, artists):
+    def _get_artist(
+        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):
+    def _get_tracks(self, query: str) -> list[TrackInfo]:
         """Returns a list of TrackInfo objects for a Beatport query."""
         bp_tracks = self.client.search(query, release_type="track")
         return [self._get_track_info(x) for x in bp_tracks]