Skip to main content
`Ctrl`+`K`

# Source code for beetsplug._utils.requests

    from __future__ import annotations

    import atexit
    import threading
    import time
    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 requests.adapters import HTTPAdapter
    from urllib3.util.retry import Retry

    from beets import __version__

    if TYPE_CHECKING:
        from collections.abc import Iterator

    class BeetsHTTPError(requests.exceptions.HTTPError):
        STATUS: ClassVar[HTTPStatus]

        def __init__(self, *args, message: str | None = None, **kwargs) -> None:
            if not message:
                message = f"HTTP Error: {self.STATUS.value} {self.STATUS.phrase}"

            super().__init__(message, *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 TimeoutAndRetrySession(requests.Session, metaclass=SingletonMeta):
        """HTTP session with sensible defaults.

        * default beets User-Agent header
        * default request timeout
        * automatic retries on transient connection or server errors
        * raises exceptions for HTTP error status codes
        """

        def __init__(self, *args, **kwargs) -> None:
            super().__init__(*args, **kwargs)
            self.headers["User-Agent"] = f"beets/{__version__} https://beets.io/"

            retry = Retry(
                total=6,
                backoff_factor=0.5,
                status_forcelist=[
                    HTTPStatus.INTERNAL_SERVER_ERROR,
                    HTTPStatus.BAD_GATEWAY,
                    HTTPStatus.SERVICE_UNAVAILABLE,
                    HTTPStatus.GATEWAY_TIMEOUT,
                    HTTPStatus.TOO_MANY_REQUESTS,
                ],
            )
            adapter = RateLimitAdapter(rate_limit=0.25, max_retries=retry)
            self.mount("https://", adapter)
            self.mount("http://", adapter)

        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 RateLimitAdapter(HTTPAdapter):
        """HTTPAdapter that enforces minimum interval between requests.

        Prevents server overload and 429 errors by sleeping when requests
        come too fast. Thread-safe via lock.

        Attributes:
            rate_limit: Minimum seconds between requests. Default 0.25 (4/sec).

        Override `_wait_time()` for custom strategies (token bucket, burst, etc.).
        """

        def __init__(self, rate_limit: float = 0.25, **kwargs):
            super().__init__(**kwargs)
            self.rate_limit = rate_limit
            self._last_request_time = 0.0
            self._lock = threading.Lock()

        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):
            with self._lock:
                elapsed = time.monotonic() - self._last_request_time
                wait = self._wait_time(elapsed)
                if wait > 0:
                    time.sleep(wait)
                self._last_request_time = time.monotonic()
            return super().send(request, *args, **kwargs)

    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.create_session`,
            :class:`RequestHandler.explicit_http_errors` or
            :class:`RequestHandler.status_to_error()` to customize behavior.

        Use

        - :class:`RequestHandler.get_json()` to get JSON response data
        - :class:`RequestHandler.get()` to get HTTP response object
        - :class:`RequestHandler.request()` to invoke arbitrary HTTP methods

        Feel free to define common methods that are used in multiple plugins.
        """

        #: List of custom exceptions to be raised for specific status codes.
        explicit_http_errors: ClassVar[list[type[BeetsHTTPError]]] = [
            HTTPNotFoundError
        ]

        def create_session(self) -> TimeoutAndRetrySession:
            """Create a new HTTP session instance.

            Can be overridden by subclasses to provide custom session types.
            """
            return TimeoutAndRetrySession()

        @cached_property
        def session(self) -> TimeoutAndRetrySession:
            return self.create_session()

        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, *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 self.session.request(*args, **kwargs)

        def get(self, *args, **kwargs) -> requests.Response:
            """Perform HTTP GET request with automatic error handling."""
            return self.request("get", *args, **kwargs)

        def put(self, *args, **kwargs) -> requests.Response:
            """Perform HTTP PUT request with automatic error handling."""
            return self.request("put", *args, **kwargs)

        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):
            """Fetch and parse JSON data from an HTTP endpoint."""
            return self.get(*args, **kwargs).json()