Make musicbrainz plugin talk to musicbrainz directly

.git-blame-ignore-revs

@@ -49,6 +49,10 @@ f36bc497c8c8f89004f3f6879908d3f0b25123e1
 c490ac5810b70f3cf5fd8649669838e8fdb19f4d
 # Importer restructure
 9147577b2b19f43ca827e9650261a86fb0450cef
+# Move functionality under MusicBrainz plugin
+529aaac7dced71266c6d69866748a7d044ec20ff
+# musicbrainz: reorder methods
+5dc6f45110b99f0cc8dbb94251f9b1f6d69583fa
 # Copy paste query, types from library to dbcore
 1a045c91668c771686f4c871c84f1680af2e944b
 # Library restructure (split library.py into multiple modules)

.github/workflows/ci.yaml

@@ -58,15 +58,15 @@ jobs:
       - if: ${{ env.IS_MAIN_PYTHON != 'true' }}
         name: Test without coverage
         run: |
-          poetry install --without=lint --extras=autobpm --extras=lyrics --extras=replaygain --extras=reflink --extras=fetchart --extras=chroma --extras=sonosupdate
+          poetry install --without=lint --extras=autobpm --extras=lyrics --extras=replaygain --extras=reflink --extras=fetchart --extras=chroma --extras=sonosupdate --extras=parentwork
           poe test
 
       - if: ${{ env.IS_MAIN_PYTHON == 'true' }}
         name: Test with coverage
         env:
           LYRICS_UPDATED: ${{ steps.lyrics-update.outputs.any_changed }}
         run: |
-          poetry install --extras=autobpm --extras=lyrics --extras=docs --extras=replaygain --extras=reflink --extras=fetchart --extras=chroma --extras=sonosupdate
+          poetry install --extras=autobpm --extras=lyrics --extras=docs --extras=replaygain --extras=reflink --extras=fetchart --extras=chroma --extras=sonosupdate --extras=parentwork
           poe docs
           poe test-with-coverage
 

beetsplug/_utils/requests.py

@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+import atexit
+import threading
+from contextlib import contextmanager
+from functools import cached_property
+from http import HTTPStatus
+from typing import TYPE_CHECKING, Any, ClassVar, Generic, Protocol, TypeVar
+
+import requests
+
+from beets import __version__
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+class BeetsHTTPError(requests.exceptions.HTTPError):
+    STATUS: ClassVar[HTTPStatus]
+
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(
+            f"HTTP Error: {self.STATUS.value} {self.STATUS.phrase}",
+            *args,
+            **kwargs,
+        )
+
+
+class HTTPNotFoundError(BeetsHTTPError):
+    STATUS = HTTPStatus.NOT_FOUND
+
+
+class Closeable(Protocol):
+    """Protocol for objects that have a close method."""
+
+    def close(self) -> None: ...
+
+
+C = TypeVar("C", bound=Closeable)
+
+
+class SingletonMeta(type, Generic[C]):
+    """Metaclass ensuring a single shared instance per class.
+
+    Creates one instance per class type on first instantiation, reusing it
+    for all subsequent calls. Automatically registers cleanup on program exit
+    for proper resource management.
+    """
+
+    _instances: ClassVar[dict[type[Any], Any]] = {}
+    _lock: ClassVar[threading.Lock] = threading.Lock()
+
+    def __call__(cls, *args: Any, **kwargs: Any) -> C:
+        if cls not in cls._instances:
+            with cls._lock:
+                if cls not in SingletonMeta._instances:
+                    instance = super().__call__(*args, **kwargs)
+                    SingletonMeta._instances[cls] = instance
+                    atexit.register(instance.close)
+        return SingletonMeta._instances[cls]
+
+
+class TimeoutSession(requests.Session, metaclass=SingletonMeta):
+    """HTTP session with automatic timeout and status checking.
+
+    Extends requests.Session to provide sensible defaults for beets HTTP
+    requests: automatic timeout enforcement, status code validation, and
+    proper user agent identification.
+    """
+
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self.headers["User-Agent"] = f"beets/{__version__} https://beets.io/"
+
+    def request(self, *args, **kwargs):
+        """Execute HTTP request with automatic timeout and status validation.
+
+        Ensures all requests have a timeout (defaults to 10 seconds) and raises
+        an exception for HTTP error status codes.
+        """
+        kwargs.setdefault("timeout", 10)
+        r = super().request(*args, **kwargs)
+        r.raise_for_status()
+
+        return r
+
+
+class RequestHandler:
+    """Manages HTTP requests with custom error handling and session management.
+
+    Provides a reusable interface for making HTTP requests with automatic
+    conversion of standard HTTP errors to beets-specific exceptions. Supports
+    custom session types and error mappings that can be overridden by
+    subclasses.
+
+    Usage:
+        Subclass and override :class:`RequestHandler.session_type`,
+        :class:`RequestHandler.explicit_http_errors` or
+        :class:`RequestHandler.status_to_error()` to customize behavior.
+
+        Use :class:`RequestHandler.get()` or
+        :class:`RequestHandler.fetch_json()` for common operations, or
+        :class:`RequestHandler.request()` for full control over HTTP methods.
+
+        Feel free to define common methods that are used in multiple plugins.
+    """
+
+    session_type: ClassVar[type[TimeoutSession]] = TimeoutSession
+    explicit_http_errors: ClassVar[list[type[BeetsHTTPError]]] = [
+        HTTPNotFoundError
+    ]
+
+    @cached_property
+    def session(self) -> Any:
+        """Lazily initialize and cache the HTTP session."""
+        return self.session_type()
+
+    def status_to_error(
+        self, code: int
+    ) -> type[requests.exceptions.HTTPError] | None:
+        """Map HTTP status codes to beets-specific exception types.
+
+        Searches the configured explicit HTTP errors for a matching status code.
+        Returns None if no specific error type is registered for the given code.
+        """
+        return next(
+            (e for e in self.explicit_http_errors if e.STATUS == code), None
+        )
+
+    @contextmanager
+    def handle_http_error(self) -> Iterator[None]:
+        """Convert standard HTTP errors to beets-specific exceptions.
+
+        Wraps operations that may raise HTTPError, automatically translating
+        recognized status codes into their corresponding beets exception types.
+        Unrecognized errors are re-raised unchanged.
+        """
+        try:
+            yield
+        except requests.exceptions.HTTPError as e:
+            if beets_error := self.status_to_error(e.response.status_code):
+                raise beets_error(response=e.response) from e
+            raise
+
+    def request(self, method: str, *args, **kwargs) -> requests.Response:
+        """Perform HTTP request using the session with automatic error handling.
+
+        Delegates to the underlying session method while converting recognized
+        HTTP errors to beets-specific exceptions through the error handler.
+        """
+        with self.handle_http_error():
+            return getattr(self.session, method)(*args, **kwargs)
+
+    def get(self, *args, **kwargs) -> requests.Response:
+        """Perform HTTP GET request with automatic error handling."""
+        return self.request("get", *args, **kwargs)
+
+    def fetch_json(self, *args, **kwargs):
+        """Fetch and parse JSON data from an HTTP endpoint."""
+        return self.get(*args, **kwargs).json()

beetsplug/lyrics.py

@@ -16,7 +16,6 @@
 
 from __future__ import annotations
 
-import atexit
 import itertools
 import math
 import re
@@ -25,23 +24,27 @@
 from dataclasses import dataclass
 from functools import cached_property, partial, total_ordering
 from html import unescape
-from http import HTTPStatus
 from itertools import groupby
 from pathlib import Path
-from typing import TYPE_CHECKING, Iterable, Iterator, NamedTuple
+from typing import TYPE_CHECKING, NamedTuple
 from urllib.parse import quote, quote_plus, urlencode, urlparse
 
 import langdetect
 import requests
 from bs4 import BeautifulSoup
 from unidecode import unidecode
 
-import beets
 from beets import plugins, ui
 from beets.autotag.distance import string_dist
 from beets.util.config import sanitize_choices
 
+from ._utils.requests import HTTPNotFoundError, RequestHandler
+
 if TYPE_CHECKING:
+    from collections.abc import Iterable, Iterator
+
+    import confuse
+
     from beets.importer import ImportTask
     from beets.library import Item, Library
     from beets.logging import BeetsLogger as Logger
@@ -54,41 +57,12 @@
         TranslatorAPI,
     )
 
-USER_AGENT = f"beets/{beets.__version__}"
 INSTRUMENTAL_LYRICS = "[Instrumental]"
 
 
-class NotFoundError(requests.exceptions.HTTPError):
-    pass
-
-
 class CaptchaError(requests.exceptions.HTTPError):
-    pass
-
-
-class TimeoutSession(requests.Session):
-    def request(self, *args, **kwargs):
-        """Wrap the request method to raise an exception on HTTP errors."""
-        kwargs.setdefault("timeout", 10)
-        r = super().request(*args, **kwargs)
-        if r.status_code == HTTPStatus.NOT_FOUND:
-            raise NotFoundError("HTTP Error: Not Found", response=r)
-        if 300 <= r.status_code < 400:
-            raise CaptchaError("Captcha is required", response=r)
-
-        r.raise_for_status()
-
-        return r
-
-
-r_session = TimeoutSession()
-r_session.headers.update({"User-Agent": USER_AGENT})
-
-
-@atexit.register
-def close_session():
-    """Close the requests session on shut down."""
-    r_session.close()
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__("Captcha is required", *args, **kwargs)
 
 
 # Utilities.
@@ -184,9 +158,18 @@ def slug(text: str) -> str:
     return re.sub(r"\W+", "-", unidecode(text).lower().strip()).strip("-")
 
 
-class RequestHandler:
+class LyricsRequestHandler(RequestHandler):
     _log: Logger
 
+    def status_to_error(self, code: int) -> type[requests.HTTPError] | None:
+        if err := super().status_to_error(code):
+            return err
+
+        if 300 <= code < 400:
+            return CaptchaError
+
+        return None
+
     def debug(self, message: str, *args) -> None:
         """Log a debug message with the class name."""
         self._log.debug(f"{self.__class__.__name__}: {message}", *args)
@@ -216,21 +199,21 @@ def fetch_text(
         """
         url = self.format_url(url, params)
         self.debug("Fetching HTML from {}", url)
-        r = r_session.get(url, **kwargs)
+        r = self.get(url, **kwargs)
         r.encoding = None
         return r.text
 
     def fetch_json(self, url: str, params: JSONDict | None = None, **kwargs):
         """Return JSON data from the given URL."""
         url = self.format_url(url, params)
         self.debug("Fetching JSON from {}", url)
-        return r_session.get(url, **kwargs).json()
+        return super().fetch_json(url, **kwargs)
 
     def post_json(self, url: str, params: JSONDict | None = None, **kwargs):
         """Send POST request and return JSON response."""
         url = self.format_url(url, params)
         self.debug("Posting JSON to {}", url)
-        return r_session.post(url, **kwargs).json()
+        return self.request("post", url, **kwargs).json()
 
     @contextmanager
     def handle_request(self) -> Iterator[None]:
@@ -249,8 +232,10 @@ def name(cls) -> str:
         return cls.__name__.lower()
 
 
-class Backend(RequestHandler, metaclass=BackendClass):
-    def __init__(self, config, log):
+class Backend(LyricsRequestHandler, metaclass=BackendClass):
+    config: confuse.Subview
+
+    def __init__(self, config: confuse.Subview, log: Logger) -> None:
         self._log = log
         self.config = config
 
@@ -356,7 +341,7 @@ def fetch_candidates(
 
         yield self.fetch_json(self.SEARCH_URL, params=base_params)
 
-        with suppress(NotFoundError):
+        with suppress(HTTPNotFoundError):
             yield [self.fetch_json(self.GET_URL, params=get_params)]
 
     @classmethod
@@ -741,7 +726,7 @@ def scrape(cls, html: str) -> str | None:
 
 
 @dataclass
-class Translator(RequestHandler):
+class Translator(LyricsRequestHandler):
     TRANSLATE_URL = "https://api.cognitive.microsofttranslator.com/translate"
     LINE_PARTS_RE = re.compile(r"^(\[\d\d:\d\d.\d\d\]|) *(.*)$")
     SEPARATOR = " | "
@@ -949,7 +934,7 @@ def write(self, items: list[Item]) -> None:
         ui.print_(textwrap.dedent(text))
 
 
-class LyricsPlugin(RequestHandler, plugins.BeetsPlugin):
+class LyricsPlugin(LyricsRequestHandler, plugins.BeetsPlugin):
     BACKEND_BY_NAME = {
         b.name: b for b in [LRCLib, Google, Genius, Tekstowo, MusiXmatch]
     }