Prepare musicbrainzngs replacement

Some of the work that was done in musicbrainzngs requires to be done in beets to
lighten the musicbrainz API.

Signed-off-by: Louis Rannou <[email protected]>

Author: Louis Rannou

beets/autotag/mb.py

@@ -49,7 +49,64 @@
     "year": "date",
 }
 
-musicbrainzngs.set_useragent("beets", beets.__version__, "https://beets.io/")
+
+class MbInterface():
+    VALID_BROWSE_INCLUDES = musicbrainzngs.VALID_BROWSE_INCLUDES
+    VALID_INCLUDES = musicbrainzngs.VALID_INCLUDES
+
+    class WebServiceError(musicbrainzngs.WebServiceError):
+        pass
+
+    class MusicBrainzError(musicbrainzngs.MusicBrainzError):
+        pass
+
+    class ResponseError (musicbrainzngs.ResponseError):
+        pass
+
+    def __init__(self):
+        pass
+
+    def set_useragent(self, app, version, contact=None):
+        return musicbrainzngs.set_useragent(app, version, contact=contact)
+
+    def set_hostname(self, hostname, https):
+        return musicbrainzngs.set_hostname(hostname, https)
+
+    def set_rate_limit(self, interval, ratelimit):
+        return musicbrainzngs.set_rate_limit(interval, ratelimit)
+
+    def browse_recordings(self, artist=None, release=None, includes=[],
+                          limit=None, offset=None):
+        return musicbrainzngs.browse_recordings(
+            artist=artist, release=release, includes=includes,
+            limit=limit, offset=offset)
+
+    def search_recordings(self, query='', limit=None, offset=None,
+                          strict=False, **fields):
+        return musicbrainzngs.search_recordings(
+            query=query, limit=limit, offset=offset,
+            strict=strict, **fields)
+
+    def get_release_by_id(self, id, includes=[],
+                          release_status=[], release_type=[]):
+        return musicbrainzngs.get_release_by_id(id, includes=includes,
+                                                release_status=release_status,
+                                                release_type=release_type)
+
+    def get_recording_by_id(self, id, includes=[],
+                            release_status=[], release_type=[]):
+        return musicbrainzngs.get_recording_by_id(id, includes=includes,
+                                                  release_status=release_status,
+                                                  release_type=release_type)
+
+    def search_releases(self, query='', limit=None, offset=None,
+                        strict=False, **fields):
+        return musicbrainzngs.search_releases(query=query, limit=limit, offset=offset,
+                                              strict=strict, **fields)
+
+
+mbi = MbInterface()
+mbi.set_useragent("beets", beets.__version__, "https://beets.io/")
 
 
 class MusicBrainzAPIError(util.HumanReadableException):
@@ -59,7 +116,7 @@ class MusicBrainzAPIError(util.HumanReadableException):
 
     def __init__(self, reason, verb, query, tb=None):
         self.query = query
-        if isinstance(reason, musicbrainzngs.WebServiceError):
+        if isinstance(reason, mbi.WebServiceError):
             reason = "MusicBrainz not reachable"
         super().__init__(reason, verb, tb)
 
@@ -94,14 +151,14 @@ def get_message(self):
     "recording-rels",
     "release-rels",
 ]
-if "work-level-rels" in musicbrainzngs.VALID_BROWSE_INCLUDES["recording"]:
+if "work-level-rels" in mbi.VALID_BROWSE_INCLUDES["recording"]:
     BROWSE_INCLUDES.append("work-level-rels")
 BROWSE_CHUNKSIZE = 100
 BROWSE_MAXTRACKS = 500
 TRACK_INCLUDES = ["artists", "aliases", "isrcs"]
-if "work-level-rels" in musicbrainzngs.VALID_INCLUDES["recording"]:
+if "work-level-rels" in mbi.VALID_INCLUDES["recording"]:
     TRACK_INCLUDES += ["work-level-rels", "artist-rels"]
-if "genres" in musicbrainzngs.VALID_INCLUDES["recording"]:
+if "genres" in mbi.VALID_INCLUDES["recording"]:
     RELEASE_INCLUDES += ["genres"]
 
 
@@ -122,8 +179,8 @@ def configure():
     # Only call set_hostname when a custom server is configured. Since
     # musicbrainz-ngs connects to musicbrainz.org with HTTPS by default
     if hostname != "musicbrainz.org":
-        musicbrainzngs.set_hostname(hostname, https)
-    musicbrainzngs.set_rate_limit(
+        mbi.set_hostname(hostname, https)
+    mbi.set_rate_limit(
         config["musicbrainz"]["ratelimit_interval"].as_number(),
         config["musicbrainz"]["ratelimit"].get(int),
     )
@@ -427,7 +484,7 @@ def album_info(release: Dict) -> beets.autotag.hooks.AlbumInfo:
         for i in range(0, ntracks, BROWSE_CHUNKSIZE):
             log.debug("Retrieving tracks starting at {}", i)
             recording_list.extend(
-                musicbrainzngs.browse_recordings(
+                mbi.browse_recordings(
                     release=release["id"],
                     limit=BROWSE_CHUNKSIZE,
                     includes=BROWSE_INCLUDES,
@@ -720,10 +777,10 @@ def match_album(
 
     try:
         log.debug("Searching for MusicBrainz releases with: {!r}", criteria)
-        res = musicbrainzngs.search_releases(
+        res = mbi.search_releases(
             limit=config["musicbrainz"]["searchlimit"].get(int), **criteria
         )
-    except musicbrainzngs.MusicBrainzError as exc:
+    except mbi.MusicBrainzError as exc:
         raise MusicBrainzAPIError(
             exc, "release search", criteria, traceback.format_exc()
         )
@@ -751,10 +808,10 @@ def match_track(
         return
 
     try:
-        res = musicbrainzngs.search_recordings(
+        res = mbi.search_recordings(
             limit=config["musicbrainz"]["searchlimit"].get(int), **criteria
         )
-    except musicbrainzngs.MusicBrainzError as exc:
+    except mbi.MusicBrainzError as exc:
         raise MusicBrainzAPIError(
             exc, "recording search", criteria, traceback.format_exc()
         )
@@ -794,7 +851,7 @@ def _find_actual_release_from_pseudo_release(
 
     actual_id = translations[0]["target"]
 
-    return musicbrainzngs.get_release_by_id(actual_id, RELEASE_INCLUDES)
+    return mbi.get_release_by_id(actual_id, RELEASE_INCLUDES)
 
 
 def _merge_pseudo_and_actual_album(
@@ -847,18 +904,18 @@ def album_for_id(releaseid: str) -> Optional[beets.autotag.hooks.AlbumInfo]:
         log.debug("Invalid MBID ({0}).", releaseid)
         return None
     try:
-        res = musicbrainzngs.get_release_by_id(albumid, RELEASE_INCLUDES)
+        res = mbi.get_release_by_id(albumid, RELEASE_INCLUDES)
 
         # resolve linked release relations
         actual_res = None
 
         if res["release"].get("status") == "Pseudo-Release":
             actual_res = _find_actual_release_from_pseudo_release(res)
 
-    except musicbrainzngs.ResponseError:
+    except mbi.ResponseError:
         log.debug("Album ID match failed.")
         return None
-    except musicbrainzngs.MusicBrainzError as exc:
+    except mbi.MusicBrainzError as exc:
         raise MusicBrainzAPIError(
             exc, "get release by ID", albumid, traceback.format_exc()
         )
@@ -883,11 +940,11 @@ def track_for_id(releaseid: str) -> Optional[beets.autotag.hooks.TrackInfo]:
         log.debug("Invalid MBID ({0}).", releaseid)
         return None
     try:
-        res = musicbrainzngs.get_recording_by_id(trackid, TRACK_INCLUDES)
-    except musicbrainzngs.ResponseError:
+        res = mbi.get_recording_by_id(trackid, TRACK_INCLUDES)
+    except mbi.ResponseError:
         log.debug("Track ID match failed.")
         return None
-    except musicbrainzngs.MusicBrainzError as exc:
+    except mbi.MusicBrainzError as exc:
         raise MusicBrainzAPIError(
             exc, "get recording by ID", trackid, traceback.format_exc()
         )

test/test_mb.py

@@ -775,7 +775,8 @@ def test_match_track(self):
             }
             ti = list(mb.match_track("hello", "there"))[0]
 
-            p.assert_called_with(artist="hello", recording="there", limit=5)
+            p.assert_called_with(query="", limit=5, offset=None, strict=False,
+                                 artist="hello", recording="there")
             self.assertEqual(ti.title, "foo")
             self.assertEqual(ti.track_id, "bar")
 
@@ -828,8 +829,10 @@ def test_match_album(self):
 
                 ai = list(mb.match_album("hello", "there"))[0]
 
-                sp.assert_called_with(artist="hello", release="there", limit=5)
-                gp.assert_called_with(mbid, mock.ANY)
+                sp.assert_called_with(query="", limit=5, offset=None, strict=False,
+                                      artist="hello", release="there")
+                gp.assert_called_with(mbid, includes=mock.ANY,
+                                      release_status=[], release_type=[])
                 self.assertEqual(ai.tracks[0].title, "foo")
                 self.assertEqual(ai.album, "hi")