music-assistant-server

1"""MusicBrainz API client."""
2
3from __future__ import annotations
4
5import logging
6from typing import TYPE_CHECKING, Any
7
8from music_assistant_models.errors import RateLimited, ResourceTemporarilyUnavailable
9
10from music_assistant.controllers.cache import use_cache
11from music_assistant.helpers.json import json_loads
12from music_assistant.helpers.throttle_retry import (
13    ThrottlerManager,
14    parse_retry_after,
15    throttle_with_retries,
16)
17
18if TYPE_CHECKING:
19    from music_assistant.mass import MusicAssistant
20
21MB_BASE_URL = "https://musicbrainz-mirror.music-assistant.io/ws/2"
22
23
24class MusicBrainzAPIClient:
25    """Thin HTTP client for the MusicBrainz API."""
26
27    domain = "musicbrainz"
28    throttler = ThrottlerManager(rate_limit=10, period=10)
29
30    def __init__(self, mass: MusicAssistant) -> None:
31        """Initialize the API client."""
32        self.mass = mass
33        self.logger = logging.getLogger(__name__)
34
35    @use_cache(86400 * 30)  # Cache for 30 days
36    @throttle_with_retries
37    async def get_data(self, endpoint: str, **kwargs: str) -> Any:
38        """
39        Fetch data from the MusicBrainz API.
40
41        Results are cached for 30 days.
42
43        :param endpoint: API endpoint path (e.g. ``artist/123?inc=aliases``).
44        :param kwargs: Additional query parameters forwarded as URL params.
45        """
46        url = f"{MB_BASE_URL}/{endpoint}"
47        headers = {
48            "User-Agent": f"Music Assistant/{self.mass.version} (https://music-assistant.io)"
49        }
50        kwargs["fmt"] = "json"
51        async with self.mass.http_session.get(url, headers=headers, params=kwargs) as response:
52            # handle rate limiter
53            if response.status == 429:
54                backoff_time = parse_retry_after(response.headers.get("Retry-After"))
55                raise RateLimited("Rate Limiter", backoff_time=backoff_time)
56            # handle temporary server error
57            if response.status in (502, 503):
58                raise ResourceTemporarilyUnavailable(backoff_time=30)
59            # handle 404 not found
60            if response.status in (400, 401, 404):
61                return None
62            response.raise_for_status()
63            return await response.json(loads=json_loads)
64