music-assistant-server

1"""Manage MediaItems of type Artist."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7from itertools import zip_longest
8from typing import TYPE_CHECKING, Any, Literal, cast, overload
9
10from music_assistant_models.auth import Scope
11from music_assistant_models.enums import (
12    AlbumType,
13    ArtistType,
14    MediaType,
15    ProviderFeature,
16    ProviderType,
17)
18from music_assistant_models.errors import (
19    MediaNotFoundError,
20    MusicAssistantError,
21    ProviderUnavailableError,
22)
23from music_assistant_models.helpers import create_safe_string
24from music_assistant_models.media_items import (
25    Album,
26    Artist,
27    ArtistSummary,
28    Audiobook,
29    ItemMapping,
30    MediaCollection,
31    ProviderMapping,
32    Track,
33)
34
35from music_assistant.constants import (
36    DB_TABLE_ALBUM_ARTISTS,
37    DB_TABLE_ARTISTS,
38    DB_TABLE_AUDIOBOOK_ARTISTS,
39    DB_TABLE_TRACK_ARTISTS,
40    VARIOUS_ARTISTS_MBID,
41    VARIOUS_ARTISTS_NAME,
42)
43from music_assistant.helpers.compare import (
44    compare_album,
45    compare_artist,
46    compare_strings,
47    compare_track,
48)
49from music_assistant.helpers.database import UNSET
50from music_assistant.helpers.json import serialize_to_json
51from music_assistant.models.music_provider import MusicProvider
52
53from .base import MediaControllerBase
54
55if TYPE_CHECKING:
56    from collections.abc import Mapping
57
58    from music_assistant import MusicAssistant
59    from music_assistant.models.metadata_provider import MetadataProvider
60
61
62class ArtistsController(MediaControllerBase[Artist]):
63    """Controller managing MediaItems of type Artist."""
64
65    db_table = DB_TABLE_ARTISTS
66    media_type = MediaType.ARTIST
67    item_cls = Artist
68    summary_item_cls = ArtistSummary
69
70    def __init__(self, mass: MusicAssistant) -> None:
71        """Initialize class."""
72        super().__init__(mass)
73        self._db_add_lock = asyncio.Lock()
74        # register (extra) api handlers
75        api_base = self.api_base
76        self.mass.register_api_command(
77            f"music/{api_base}/artist_albums", self.albums, required_scope=Scope.LIBRARY_READ
78        )
79        self.mass.register_api_command(
80            f"music/{api_base}/artist_tracks", self.tracks, required_scope=Scope.LIBRARY_READ
81        )
82        self.mass.register_api_command(
83            f"music/{api_base}/top_tracks", self.top_tracks, required_scope=Scope.LIBRARY_READ
84        )
85        self.mass.register_api_command(
86            f"music/{api_base}/top_albums", self.top_albums, required_scope=Scope.LIBRARY_READ
87        )
88        self.mass.register_api_command(
89            f"music/{api_base}/artist_audiobooks",
90            self.audiobooks,
91            required_scope=Scope.LIBRARY_READ,
92        )
93        self.mass.register_api_command(
94            f"music/{api_base}/similar_artists",
95            self.similar_artists,
96            required_scope=Scope.LIBRARY_READ,
97        )
98        self.mass.register_api_command(
99            f"music/{api_base}/library_artist_types",
100            self.get_library_artist_types,
101            required_scope=Scope.LIBRARY_READ,
102        )
103
104    @property
105    def summary_query(self) -> tuple[str, dict[str, Any]]:
106        """Return the slim SELECT query used for artist summary listings."""
107        query = f"""
108        SELECT
109            {self._summary_base_columns()},
110            artists.artist_type,
111            {self._provider_mappings_query()} AS provider_mappings
112            FROM artists"""
113        return query, {}
114
115    async def library_count(
116        self,
117        favorite_only: bool = False,
118        album_artists_only: bool = False,
119        artist_type: ArtistType | None = None,
120    ) -> int:
121        """
122        Return the number of artists in the library.
123
124        Restricted to the providers the current user is allowed to see when that user
125        has a provider filter set.
126
127        :param favorite_only: Only count artists marked as favorite.
128        :param album_artists_only: Only count artists that have albums.
129        :param artist_type: Only count artists of this type.
130        """
131        sql_query = f"SELECT item_id FROM {self.db_table}"
132        query_parts = []
133        query_params: dict[str, Any] = {}
134        if artist_type:
135            query_parts.append(f"artist_type = '{artist_type}'")
136        if favorite_only:
137            query_parts.append("favorite = 1")
138        if album_artists_only:
139            query_parts.append(
140                f"item_id in (select {DB_TABLE_ALBUM_ARTISTS}.artist_id "
141                f"FROM {DB_TABLE_ALBUM_ARTISTS})"
142            )
143        if provider_filter := self._ensure_provider_filter(None):
144            query_parts.append(
145                self._provider_filter_clause(query_params, provider_filter, in_library_only=True)
146            )
147        if query_parts:
148            sql_query += f" WHERE {' AND '.join(query_parts)}"
149        return await self.mass.music.database.get_count_from_query(sql_query, query_params)
150
151    async def library_items(  # noqa: PLR0913
152        self,
153        favorite: bool | None = None,
154        search: str | None = None,
155        limit: int = 500,
156        offset: int = 0,
157        order_by: str = "sort_name",
158        provider: str | list[str] | None = None,
159        genre: int | list[int] | None = None,
160        played_only: bool = False,
161        album_artists_only: bool = False,
162        artist_type: ArtistType | None = None,
163        *,
164        summary: bool = True,
165        reachable_via: list[str] | None = None,
166        **kwargs: Any,
167    ) -> list[Artist]:
168        """
169        Get in-database (album) artists.
170
171        :param favorite: Filter by favorite status.
172        :param search: Filter by search query.
173        :param limit: Maximum number of items to return.
174        :param offset: Number of items to skip.
175        :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
176        :param provider: Filter by provider instance ID (single string or list).
177        :param album_artists_only: Only return artists that have albums.
178        :param genre: Filter by genre id(s).
179        :param artist_type: The artist's type
180        :param summary: When True (default), return slim summary items containing only the
181            fields needed for a list view. Set to False to get fully hydrated items.
182        :param reachable_via: Restrict results to items with a provider mapping reachable
183            through one of these provider instance ids (OR semantics). See
184            `MediaControllerBase.library_items` for the full semantics.
185        """
186        reachable_via = self._resolve_reachable_via(reachable_via)
187        if reachable_via is not None and not reachable_via:
188            return []
189        extra_query_params: dict[str, Any] = {}
190        extra_query_parts: list[str] = []
191        if artist_type:
192            extra_query_parts = [f"artist_type = '{artist_type}'"]
193        if album_artists_only and artist_type in (None, ArtistType.SINGER):
194            extra_query_parts.append(
195                f"artists.item_id in (select {DB_TABLE_ALBUM_ARTISTS}.artist_id "
196                f"from {DB_TABLE_ALBUM_ARTISTS})"
197            )
198        return await self.get_library_items_by_query(
199            favorite=favorite,
200            search=search,
201            genre_ids=genre,
202            limit=limit,
203            offset=offset,
204            order_by=order_by,
205            provider_filter=self._provider_filter_considering_reachability(provider, reachable_via),
206            extra_query_parts=extra_query_parts,
207            extra_query_params=extra_query_params,
208            played_only=played_only,
209            in_library_only=True,
210            summary=summary,
211            reachable_via=reachable_via,
212        )
213
214    async def tracks(
215        self,
216        item_id: str,
217        provider_instance_id_or_domain: str,
218        provider_filter: str | None = None,
219    ) -> list[Track]:
220        """
221        Return the tracks for a artist.
222
223        For a library item, the in-library tracks are returned, optionally limited to a single
224        provider instance with the provider_filter. For a provider item, that provider's
225        tracks listing is returned (which may be empty if it is not supported).
226
227        :param item_id: The item ID of the artist.
228        :param provider_instance_id_or_domain: The provider instance ID or domain of the artist.
229        :param provider_filter: Optional provider instance ID to limit the (library) result to.
230        """
231        if provider_instance_id_or_domain == "library":
232            return await self.get_library_artist_tracks(item_id, provider_filter=provider_filter)
233        self._validate_provider_filter(provider_instance_id_or_domain, provider_filter)
234        return await self.get_provider_artist_tracks(item_id, provider_instance_id_or_domain)
235
236    async def albums(
237        self,
238        item_id: str,
239        provider_instance_id_or_domain: str,
240        provider_filter: str | None = None,
241    ) -> list[Album]:
242        """
243        Return the albums for an artist.
244
245        For a library item, the in-library albums are returned, optionally limited to a single
246        provider instance with the provider_filter. For a provider item, that provider's
247        albums listing is returned (which may be empty if it is not supported).
248
249        :param item_id: The item ID of the artist.
250        :param provider_instance_id_or_domain: The provider instance ID or domain of the artist.
251        :param provider_filter: Optional provider instance ID to limit the (library) result to.
252        """
253        if provider_instance_id_or_domain == "library":
254            return await self.get_library_artist_albums(item_id, provider_filter=provider_filter)
255        self._validate_provider_filter(provider_instance_id_or_domain, provider_filter)
256        return await self.get_provider_artist_albums(item_id, provider_instance_id_or_domain)
257
258    async def top_tracks(
259        self,
260        item_id: str,
261        provider_instance_id_or_domain: str,
262        provider_filter: str | None = None,
263    ) -> list[Track]:
264        """
265        Return the top/featured tracks for an artist.
266
267        For a library item, the top tracks of all the artist's providers are aggregated (and
268        deduplicated), optionally limited to a single provider instance. For a provider
269        item, that provider's top tracks listing is returned (may be empty if not supported).
270
271        :param item_id: The item ID of the artist.
272        :param provider_instance_id_or_domain: The provider instance ID or domain of the artist.
273        :param provider_filter: Optional provider instance ID to limit the result to.
274        """
275        if provider_instance_id_or_domain == "library":
276            return await self.get_library_artist_toptracks(item_id, provider_filter=provider_filter)
277        self._validate_provider_filter(provider_instance_id_or_domain, provider_filter)
278        return await self.get_provider_artist_toptracks(item_id, provider_instance_id_or_domain)
279
280    async def top_albums(
281        self,
282        item_id: str,
283        provider_instance_id_or_domain: str,
284        provider_filter: str | None = None,
285    ) -> list[Album]:
286        """
287        Return the top/featured albums for an artist.
288
289        For a library item, the top albums of all the artist's providers are aggregated (and
290        deduplicated), optionally limited to a single provider instance. For a provider
291        item, that provider's top albums listing is returned (may be empty if not supported).
292
293        :param item_id: The item ID of the artist.
294        :param provider_instance_id_or_domain: The provider instance ID or domain of the artist.
295        :param provider_filter: Optional provider instance ID to limit the result to.
296        """
297        if provider_instance_id_or_domain == "library":
298            return await self.get_library_artist_topalbums(item_id, provider_filter=provider_filter)
299        self._validate_provider_filter(provider_instance_id_or_domain, provider_filter)
300        return await self.get_provider_artist_topalbums(item_id, provider_instance_id_or_domain)
301
302    async def similar_artists(
303        self,
304        item_id: str,
305        provider_instance_id_or_domain: str,
306        provider_filter: str | None = None,
307        limit: int = 25,
308    ) -> list[Artist]:
309        """
310        Return similar artists for an artist.
311
312        For a library item, the similar artists of all the artist's providers are aggregated
313        (and deduplicated), optionally limited to a single provider instance. For a provider
314        item, that provider's similar artists listing is returned (may be empty if not
315        supported).
316
317        :param item_id: The item ID of the artist.
318        :param provider_instance_id_or_domain: The provider instance ID or domain of the artist.
319        :param provider_filter: Optional provider instance ID to limit the result to.
320        :param limit: Maximum number of similar artists to return.
321        """
322        if provider_instance_id_or_domain == "library":
323            return await self.get_library_artist_similar_artists(
324                item_id, provider_filter=provider_filter, limit=limit
325            )
326        self._validate_provider_filter(provider_instance_id_or_domain, provider_filter)
327        return await self.get_provider_artist_similar_artists(
328            item_id, provider_instance_id_or_domain, limit=limit
329        )
330
331    if TYPE_CHECKING:
332
333        @overload
334        async def audiobooks(
335            self,
336            item_id: str,
337            provider_instance_id_or_domain: str,
338            artist_type: ArtistType = ArtistType.AUTHOR,
339            in_library_only: bool = False,
340            *,
341            collapse_collections: Literal[False] = False,
342        ) -> list[Audiobook]: ...
343
344        @overload
345        async def audiobooks(
346            self,
347            item_id: str,
348            provider_instance_id_or_domain: str,
349            artist_type: ArtistType = ArtistType.AUTHOR,
350            in_library_only: bool = False,
351            *,
352            collapse_collections: Literal[True],
353        ) -> list[Audiobook | MediaCollection[Audiobook]]: ...
354
355    async def audiobooks(
356        self,
357        item_id: str,
358        provider_instance_id_or_domain: str,
359        artist_type: ArtistType = ArtistType.AUTHOR,
360        in_library_only: bool = False,
361        *,
362        collapse_collections: bool = False,
363    ) -> list[Audiobook] | list[Audiobook | MediaCollection[Audiobook]]:
364        """
365        Return audiobooks for an artist.
366
367        Artist_type can be omitted for in-library artists.
368
369        :param collapse_collections: Collapse available collections. Only applies to
370            in-library items; when in_library_only is False, provider items are
371            appended as plain audiobooks alongside the collapsed collections.
372        """
373        if artist_type == ArtistType.SINGER:
374            self.logger.warning("Audiobooks not supported for artist_type SINGER.")
375            return []
376        # always check if we have a library item for this artist
377        library_artist = await self.get_library_item_by_prov_id(
378            item_id, provider_instance_id_or_domain
379        )
380        if library_artist and library_artist.artist_type == ArtistType.SINGER:
381            self.logger.debug(
382                "Ignoring audiobook request for artist of type %s", library_artist.artist_type
383            )
384            return []
385        if not library_artist:
386            if artist_type == ArtistType.AUTHOR:
387                return await self.get_provider_author_audiobooks(
388                    item_id, provider_instance_id_or_domain
389                )
390            if artist_type == ArtistType.NARRATOR:
391                return await self.get_provider_narrator_audiobooks(
392                    item_id, provider_instance_id_or_domain
393                )
394            return []
395
396        db_items = await self.get_library_author_narrator_audiobooks(
397            library_artist.item_id,
398            artist_type=library_artist.artist_type,
399            collapse_collections=collapse_collections,
400        )
401        result: list[Audiobook] | list[Audiobook | MediaCollection[Audiobook]] = db_items
402        if in_library_only:
403            # return in-library items only
404            return result
405        # return all (unique) items from all providers
406        # initialize unique_ids with db_items to prevent duplicates
407        unique_ids: set[str] = set()
408        for item in db_items:
409            if isinstance(item, MediaCollection):
410                for collection_item in item.items:
411                    unique_ids.add(f"{collection_item.name}.{collection_item.version}")
412            else:
413                unique_ids.add(f"{item.name}.{item.version}")
414        unique_providers = self.mass.music.get_unique_providers()
415        audiobook_method = (
416            self.get_provider_author_audiobooks
417            if artist_type == ArtistType.AUTHOR
418            else self.get_provider_narrator_audiobooks
419        )
420        for provider_mapping in library_artist.provider_mappings:
421            if provider_mapping.provider_instance not in unique_providers:
422                continue
423            provider_audiobooks = await audiobook_method(
424                provider_mapping.item_id, provider_mapping.provider_instance
425            )
426            for provider_audiobook in provider_audiobooks:
427                unique_id = f"{provider_audiobook.name}.{provider_audiobook.version}"
428                if unique_id in unique_ids:
429                    continue
430                unique_ids.add(unique_id)
431                # prefer db item
432                if db_item := await self.mass.music.audiobooks.get_library_item_by_prov_id(
433                    provider_audiobook.item_id, provider_audiobook.provider
434                ):
435                    result.append(db_item)
436                elif not in_library_only:
437                    result.append(provider_audiobook)
438        return result
439
440    async def get_library_author_narrator_audiobooks(
441        self,
442        item_id: str | int,
443        artist_type: ArtistType,
444        *,
445        collapse_collections: bool = False,
446    ) -> list[Audiobook] | list[Audiobook | MediaCollection[Audiobook]]:
447        """Return all in-library audiobooks for an author/ narrator."""
448        db_id = int(item_id)  # ensure integer
449        library_item = await self.get_library_item(db_id)
450        if library_item.artist_type != artist_type:
451            self.logger.debug("Audiobooks only available for artists of type %s", artist_type)
452            return []
453        subquery = (
454            f"SELECT audiobook_id FROM {DB_TABLE_AUDIOBOOK_ARTISTS} WHERE artist_id = :artist_id"
455        )
456        query = f"audiobooks.item_id in ({subquery})"
457        return await self.mass.music.audiobooks.get_library_items_by_query(
458            extra_query_parts=[query],
459            extra_query_params={"artist_id": db_id},
460            collapse_collections=collapse_collections,
461        )
462
463    async def get_provider_author_audiobooks(
464        self,
465        item_id: str,
466        provider_instance_id_or_domain: str,
467    ) -> list[Audiobook]:
468        """Return audiobooks for an author on given provider."""
469        assert provider_instance_id_or_domain != "library"
470        if not (prov := self.mass.get_provider(provider_instance_id_or_domain)):
471            return []
472        prov = cast("MusicProvider", prov)
473        if ProviderFeature.AUTHOR_AUDIOBOOKS in prov.supported_features:
474            return await prov.get_author_audiobooks(item_id)
475        # fallback implementation using the db
476        return await self._get_db_author_narrator_audiobooks(
477            item_id=item_id,
478            provider_instance_id_or_domain=provider_instance_id_or_domain,
479            artist_type=ArtistType.AUTHOR,
480        )
481
482    async def get_provider_narrator_audiobooks(
483        self,
484        item_id: str,
485        provider_instance_id_or_domain: str,
486    ) -> list[Audiobook]:
487        """Return audiobooks for an author on given provider."""
488        assert provider_instance_id_or_domain != "library"
489        if not (prov := self.mass.get_provider(provider_instance_id_or_domain)):
490            return []
491        prov = cast("MusicProvider", prov)
492        if ProviderFeature.NARRATOR_AUDIOBOOKS in prov.supported_features:
493            return await prov.get_narrator_audiobooks(item_id)
494        # fallback implementation using the db
495        return await self._get_db_author_narrator_audiobooks(
496            item_id=item_id,
497            provider_instance_id_or_domain=provider_instance_id_or_domain,
498            artist_type=ArtistType.NARRATOR,
499        )
500
501    async def get_provider_artist_toptracks(
502        self,
503        item_id: str,
504        provider_instance_id_or_domain: str,
505    ) -> list[Track]:
506        """
507        Return the top tracks for an artist on the given provider.
508
509        Each track is resolved to its in-library equivalent where available.
510        """
511        provider = self.mass.get_provider(
512            provider_instance_id_or_domain, provider_type=MusicProvider
513        )
514        if provider is None or not provider.available:
515            return []  # guard against unavailable provider
516        if not provider.supports_feature(ProviderFeature.ARTIST_TOPTRACKS):
517            self.logger.warning(
518                "Provider %s does not support fetching artist top tracks.",
519                provider.name,
520            )
521            return []  # guard against unsupported feature
522        tracks = await provider.get_artist_toptracks(item_id)
523        # resolve to in-library equivalents (in parallel) where available
524        resolved = await asyncio.gather(
525            *(
526                self.mass.music.tracks.get_library_item_by_prov_id(track.item_id, track.provider)
527                for track in tracks
528            )
529        )
530        return [
531            library_track or track for library_track, track in zip(resolved, tracks, strict=True)
532        ]
533
534    async def get_library_artist_toptracks(
535        self,
536        item_id: str | int,
537        provider_filter: str | None = None,
538    ) -> list[Track]:
539        """
540        Return the top tracks for an in-library artist, aggregated across all its providers.
541
542        The result combines (and deduplicates, preserving order) the top tracks from every
543        provider attached to the artist and any metadata/plugin provider implementing the
544        feature. Empty when no provider yields a result.
545
546        :param item_id: The library item ID of the artist.
547        :param provider_filter: Optional provider instance ID to limit the result to.
548        """
549        ref_item = await self.get_library_item(item_id)
550        allowed = self._ensure_provider_filter(provider_filter)
551        # fetch each provider's ranked top tracks in parallel
552        fetches = []
553        # streaming providers attached to the artist (results resolved to library items)
554        for provider_mapping in ref_item.provider_mappings:
555            if allowed is not None and provider_mapping.provider_instance not in allowed:
556                continue
557            music_prov = self.mass.get_provider(
558                provider_mapping.provider_instance, provider_type=MusicProvider
559            )
560            if (
561                music_prov is None
562                or ProviderFeature.ARTIST_TOPTRACKS not in music_prov.supported_features
563            ):
564                continue
565            fetches.append(
566                self.get_provider_artist_toptracks(
567                    provider_mapping.item_id, provider_mapping.provider_instance
568                )
569            )
570        # metadata/plugin providers implementing the feature
571        for prov in self.mass.get_providers_supporting_feature(
572            ProviderFeature.ARTIST_TOPTRACKS,
573            priority=(ProviderType.METADATA, ProviderType.PLUGIN),
574        ):
575            if allowed is not None and prov.instance_id not in allowed:
576                continue
577            fetches.append(cast("MetadataProvider", prov).get_artist_toptracks(ref_item))
578        per_provider = await asyncio.gather(*fetches, return_exceptions=True)
579        # drop (and log) any provider that failed so one bad provider can't sink the listing
580        listings: list[list[Track]] = []
581        for listing in per_provider:
582            if isinstance(listing, BaseException):
583                self.logger.warning(
584                    "Error fetching top tracks for artist %s from a provider",
585                    ref_item.name,
586                    exc_info=listing,
587                )
588                continue
589            listings.append(listing)
590        # interleave the providers' rankings by position (zip), deduplicating with the compare
591        # helper (which also matches on version/duration)
592        result: list[Track] = []
593        for row in zip_longest(*listings):
594            for candidate in row:
595                if candidate is None or any(
596                    compare_track(existing, candidate) for existing in result
597                ):
598                    continue
599                result.append(candidate)
600        return result
601
602    async def get_provider_artist_topalbums(
603        self,
604        item_id: str,
605        provider_instance_id_or_domain: str,
606    ) -> list[Album]:
607        """
608        Return the top/featured albums for an artist on the given provider.
609
610        Each album is resolved to its in-library equivalent where available.
611        """
612        provider = self.mass.get_provider(
613            provider_instance_id_or_domain, provider_type=MusicProvider
614        )
615        if provider is None or not provider.available:
616            return []  # guard against unavailable provider
617        if not provider.supports_feature(ProviderFeature.ARTIST_TOPALBUMS):
618            self.logger.warning(
619                "Provider %s does not support fetching artist top albums.",
620                provider.name,
621            )
622            return []  # guard against unsupported feature
623        albums = await provider.get_artist_topalbums(item_id)
624        # resolve to in-library equivalents (in parallel) where available
625        resolved = await asyncio.gather(
626            *(
627                self.mass.music.albums.get_library_item_by_prov_id(album.item_id, album.provider)
628                for album in albums
629            )
630        )
631        return [
632            library_album or album for library_album, album in zip(resolved, albums, strict=True)
633        ]
634
635    async def get_library_artist_topalbums(
636        self,
637        item_id: str | int,
638        provider_filter: str | None = None,
639    ) -> list[Album]:
640        """
641        Return the top albums for an in-library artist, aggregated across all its providers.
642
643        The result combines (and deduplicates, preserving order) the top albums from every
644        provider attached to the artist and any metadata/plugin provider implementing the
645        feature. Empty when no provider yields a result.
646
647        :param item_id: The library item ID of the artist.
648        :param provider_filter: Optional provider instance ID to limit the result to.
649        """
650        ref_item = await self.get_library_item(item_id)
651        allowed = self._ensure_provider_filter(provider_filter)
652        # fetch each provider's ranked top albums in parallel
653        fetches = []
654        # streaming providers attached to the artist (results resolved to library items)
655        for provider_mapping in ref_item.provider_mappings:
656            if allowed is not None and provider_mapping.provider_instance not in allowed:
657                continue
658            music_prov = self.mass.get_provider(
659                provider_mapping.provider_instance, provider_type=MusicProvider
660            )
661            if (
662                music_prov is None
663                or ProviderFeature.ARTIST_TOPALBUMS not in music_prov.supported_features
664            ):
665                continue
666            fetches.append(
667                self.get_provider_artist_topalbums(
668                    provider_mapping.item_id, provider_mapping.provider_instance
669                )
670            )
671        # metadata/plugin providers implementing the feature
672        for prov in self.mass.get_providers_supporting_feature(
673            ProviderFeature.ARTIST_TOPALBUMS,
674            priority=(ProviderType.METADATA, ProviderType.PLUGIN),
675        ):
676            if allowed is not None and prov.instance_id not in allowed:
677                continue
678            fetches.append(cast("MetadataProvider", prov).get_artist_topalbums(ref_item))
679        per_provider = await asyncio.gather(*fetches, return_exceptions=True)
680        # drop (and log) any provider that failed so one bad provider can't sink the listing
681        listings: list[list[Album]] = []
682        for listing in per_provider:
683            if isinstance(listing, BaseException):
684                self.logger.warning(
685                    "Error fetching top albums for artist %s from a provider",
686                    ref_item.name,
687                    exc_info=listing,
688                )
689                continue
690            listings.append(listing)
691        # interleave the providers' rankings by position (zip), deduplicating with the compare
692        # helper (which also matches on version/duration)
693        result: list[Album] = []
694        for row in zip_longest(*listings):
695            for candidate in row:
696                if candidate is None or any(
697                    compare_album(existing, candidate) for existing in result
698                ):
699                    continue
700                result.append(candidate)
701        return result
702
703    async def get_provider_artist_tracks(
704        self,
705        item_id: str,
706        provider_instance_id_or_domain: str,
707    ) -> list[Track]:
708        """Return all tracks for an artist on given provider."""
709        provider = self.mass.get_provider(
710            provider_instance_id_or_domain, provider_type=MusicProvider
711        )
712        if provider is None or not provider.available:
713            return []  # guard against unavailable provider
714        if provider.supports_feature(ProviderFeature.ARTIST_TRACKS):
715            return await provider.get_artist_tracks(item_id)
716        # fallback: enumerate (and dedupe) the tracks of all the artist's albums on the provider
717        result: list[Track] = []
718        unique_ids: set[str] = set()
719        for album in await self.get_provider_artist_albums(item_id, provider_instance_id_or_domain):
720            for track in await self.mass.music.albums.tracks(album.item_id, album.provider):
721                unique_id = f"{track.name}.{track.version}"
722                if unique_id in unique_ids:
723                    continue
724                unique_ids.add(unique_id)
725                result.append(track)
726        return result
727
728    async def get_library_artist_tracks(
729        self,
730        item_id: str | int,
731        provider_filter: str | None = None,
732    ) -> list[Track]:
733        """Return all in-library tracks for an artist, optionally limited to a single provider."""
734        db_id = int(item_id)  # ensure integer
735        library_item = await self.get_library_item(db_id)
736        if library_item.artist_type != ArtistType.SINGER:
737            self.logger.debug("Tracks only available for artists of type ARTIST")
738            return []
739        subquery = f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = :artist_id"
740        query = f"tracks.item_id in ({subquery})"
741        return await self.mass.music.tracks.get_library_items_by_query(
742            extra_query_parts=[query],
743            extra_query_params={"artist_id": db_id},
744            provider_filter=self._ensure_provider_filter(provider_filter),
745            in_library_only=True,
746        )
747
748    async def get_provider_artist_albums(
749        self,
750        item_id: str,
751        provider_instance_id_or_domain: str,
752    ) -> list[Album]:
753        """Return albums for an artist on given provider."""
754        provider = self.mass.get_provider(
755            provider_instance_id_or_domain, provider_type=MusicProvider
756        )
757        if provider is None or not provider.available:
758            return []  # guard against unavailable provider
759        if not provider.supports_feature(ProviderFeature.ARTIST_ALBUMS):
760            self.logger.warning(
761                "Provider %s does not support fetching all artist albums.",
762                provider.name,
763            )
764            return []  # guard against unsupported feature
765        return await provider.get_artist_albums(item_id)
766
767    async def get_library_artist_albums(
768        self,
769        item_id: str | int,
770        provider_filter: str | None = None,
771    ) -> list[Album]:
772        """Return all in-library albums for an artist, optionally limited to a single provider."""
773        db_id = int(item_id)  # ensure integer
774        library_item = await self.get_library_item(db_id)
775        if library_item.artist_type != ArtistType.SINGER:
776            self.logger.debug("Albums only available for artists of type ARTIST")
777            return []
778        subquery = f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = :artist_id"
779        query = f"albums.item_id in ({subquery})"
780        return await self.mass.music.albums.get_library_items_by_query(
781            extra_query_parts=[query],
782            extra_query_params={"artist_id": db_id},
783            provider_filter=self._ensure_provider_filter(provider_filter),
784            in_library_only=True,
785        )
786
787    async def get_provider_artist_similar_artists(
788        self,
789        item_id: str,
790        provider_instance_id_or_domain: str,
791        limit: int = 25,
792    ) -> list[Artist]:
793        """
794        Return similar artists for an artist on the given provider.
795
796        Each artist is resolved to its in-library equivalent where available.
797        """
798        provider = self.mass.get_provider(
799            provider_instance_id_or_domain, provider_type=MusicProvider
800        )
801        if provider is None or not provider.available:
802            return []  # guard against unavailable provider
803        if not provider.supports_feature(ProviderFeature.SIMILAR_ARTISTS):
804            self.logger.warning(
805                "Provider %s does not support fetching similar artists.",
806                provider.name,
807            )
808            return []  # guard against unsupported feature
809        artists = await provider.get_similar_artists(item_id, limit=limit)
810        # resolve to in-library equivalents (in parallel) where available
811        resolved = await asyncio.gather(
812            *(
813                self.get_library_item_by_prov_id(artist.item_id, artist.provider)
814                for artist in artists
815            )
816        )
817        return [
818            library_artist or artist
819            for library_artist, artist in zip(resolved, artists, strict=True)
820        ]
821
822    async def get_library_artist_similar_artists(
823        self,
824        item_id: str | int,
825        provider_filter: str | None = None,
826        limit: int = 25,
827    ) -> list[Artist]:
828        """
829        Return similar artists for an in-library artist, aggregated across all its providers.
830
831        The result combines (and deduplicates, preserving order) the similar artists from
832        every provider attached to the artist and any metadata/plugin provider implementing
833        the feature. Empty when no provider yields a result.
834
835        :param item_id: The library item ID of the artist.
836        :param provider_filter: Optional provider instance ID to limit the result to.
837        :param limit: Maximum number of similar artists to return.
838        """
839        ref_item = await self.get_library_item(item_id)
840        allowed = self._ensure_provider_filter(provider_filter)
841        # fetch each provider's similar artists in parallel
842        fetches = []
843        # streaming providers attached to the artist (results resolved to library items)
844        for provider_mapping in ref_item.provider_mappings:
845            if allowed is not None and provider_mapping.provider_instance not in allowed:
846                continue
847            music_prov = self.mass.get_provider(
848                provider_mapping.provider_instance, provider_type=MusicProvider
849            )
850            if (
851                music_prov is None
852                or ProviderFeature.SIMILAR_ARTISTS not in music_prov.supported_features
853            ):
854                continue
855            fetches.append(
856                self.get_provider_artist_similar_artists(
857                    provider_mapping.item_id, provider_mapping.provider_instance, limit=limit
858                )
859            )
860        # metadata/plugin providers implementing the feature
861        for prov in self.mass.get_providers_supporting_feature(
862            ProviderFeature.SIMILAR_ARTISTS,
863            priority=(ProviderType.METADATA, ProviderType.PLUGIN),
864        ):
865            if allowed is not None and prov.instance_id not in allowed:
866                continue
867            fetches.append(
868                cast("MetadataProvider", prov).get_similar_artists(ref_item, limit=limit)
869            )
870        per_provider = await asyncio.gather(*fetches, return_exceptions=True)
871        # drop (and log) any provider that failed so one bad provider can't sink the listing
872        listings: list[list[Artist]] = []
873        for listing in per_provider:
874            if isinstance(listing, BaseException):
875                self.logger.warning(
876                    "Error fetching similar artists for %s from a provider",
877                    ref_item.name,
878                    exc_info=listing,
879                )
880                continue
881            listings.append(listing)
882        # interleave the providers' results by position (zip), deduplicating with the compare
883        # helper, and cap to the requested limit
884        result: list[Artist] = []
885        for row in zip_longest(*listings):
886            for candidate in row:
887                if candidate is None or any(
888                    compare_artist(existing, candidate) for existing in result
889                ):
890                    continue
891                result.append(candidate)
892        return result[:limit]
893
894    async def get_library_artist_types(self) -> list[ArtistType]:
895        """Get all supported in-library artist types."""
896        artist_types: list[ArtistType] = []
897        query = f"SELECT DISTINCT artist_type FROM {DB_TABLE_ARTISTS}"
898        rows = await self.mass.music.database.get_rows_from_query(query)
899        for row in rows:
900            artist_types.append(ArtistType(row["artist_type"]))
901        return artist_types
902
903    async def remove_item_from_library(self, item_id: str | int, recursive: bool = True) -> None:
904        """Delete record from the database."""
905        db_id = int(item_id)  # ensure integer
906        library_item = await self.get_library_item(db_id)
907
908        if library_item.artist_type == ArtistType.SINGER:
909            await self._remove_music_artist_from_library(db_id=db_id, recursive=recursive)
910        elif library_item.artist_type in (ArtistType.AUTHOR, ArtistType.NARRATOR):
911            await self._remove_author_narrator_from_library(db_id=db_id, recursive=recursive)
912        else:
913            raise MusicAssistantError(f"Unknown artist_type {library_item.artist_type}.")
914
915        # delete the artist itself from db
916        # this will raise if the item still has references and recursive is false
917        await super().remove_item_from_library(db_id)
918
919    async def match_provider(
920        self, db_artist: Artist, provider: MusicProvider, strict: bool = True
921    ) -> list[ProviderMapping]:
922        """
923        Try to find match on (streaming) provider for the provided (database) artist.
924
925        This is used to link objects of different providers/qualities together.
926        """
927        self.logger.debug("Trying to match artist %s on provider %s", db_artist.name, provider.name)
928        matches: list[ProviderMapping] = []
929        # try to get a match with some reference tracks of this artist
930        ref_tracks = await self.mass.music.artists.tracks(db_artist.item_id, db_artist.provider)
931        if len(ref_tracks) < 10:
932            # fetch reference tracks from provider(s) attached to the artist
933            for provider_mapping in db_artist.provider_mappings:
934                with contextlib.suppress(ProviderUnavailableError, MediaNotFoundError):
935                    ref_tracks += await self.mass.music.artists.tracks(
936                        provider_mapping.item_id, provider_mapping.provider_instance
937                    )
938        for ref_track in ref_tracks:
939            search_str = f"{db_artist.name} - {ref_track.name}"
940            search_results = await self.mass.music.tracks.search(search_str, provider.domain)
941            for search_result_item in search_results:
942                if not compare_strings(search_result_item.name, ref_track.name, strict=strict):
943                    continue
944                # get matching artist from track
945                for search_item_artist in search_result_item.artists:
946                    if not compare_strings(search_item_artist.name, db_artist.name, strict=strict):
947                        continue
948                    # 100% track match
949                    # get full artist details so we have all metadata
950                    prov_artist = await self.get_provider_item(
951                        search_item_artist.item_id,
952                        search_item_artist.provider,
953                        fallback=search_item_artist,
954                    )
955                    # 100% match
956                    matches.extend(prov_artist.provider_mappings)
957                    if matches:
958                        return matches
959        # try to get a match with some reference albums of this artist
960        ref_albums = await self.mass.music.artists.albums(db_artist.item_id, db_artist.provider)
961        if len(ref_albums) < 10:
962            # fetch reference albums from provider(s) attached to the artist
963            for provider_mapping in db_artist.provider_mappings:
964                with contextlib.suppress(ProviderUnavailableError, MediaNotFoundError):
965                    ref_albums += await self.mass.music.artists.albums(
966                        provider_mapping.item_id, provider_mapping.provider_instance
967                    )
968        for ref_album in ref_albums:
969            if ref_album.album_type == AlbumType.COMPILATION:
970                continue
971            if not ref_album.artists:
972                continue
973            search_str = f"{db_artist.name} - {ref_album.name}"
974            search_result_albums = await self.mass.music.albums.search(search_str, provider.domain)
975            for search_result_album in search_result_albums:
976                if not search_result_album.artists:
977                    continue
978                if not compare_strings(search_result_album.name, ref_album.name, strict=strict):
979                    continue
980                # artist must match 100%
981                if not compare_artist(db_artist, search_result_album.artists[0], strict=strict):
982                    continue
983                # 100% match
984                # get full artist details so we have all metadata
985                prov_artist = await self.get_provider_item(
986                    search_result_album.artists[0].item_id,
987                    search_result_album.artists[0].provider,
988                    fallback=search_result_album.artists[0],
989                )
990                matches.extend(prov_artist.provider_mappings)
991                if matches:
992                    return matches
993        if not matches:
994            self.logger.debug(
995                "Could not find match for Artist %s on provider %s",
996                db_artist.name,
997                provider.name,
998            )
999        return matches
1000
1001    async def match_providers(self, db_artist: Artist) -> None:
1002        """
1003        Try to find matching artists on all providers for the provided (database) item_id.
1004
1005        This is used to link objects of different providers together.
1006        """
1007        if db_artist.provider != "library":
1008            return  # Matching only supported for database items
1009
1010        # try to find match on all providers
1011
1012        cur_provider_domains = {
1013            x.provider_domain for x in db_artist.provider_mappings if x.available
1014        }
1015        for provider in self.mass.music.providers:
1016            if provider.domain in cur_provider_domains:
1017                continue
1018            if ProviderFeature.SEARCH not in provider.supported_features:
1019                continue
1020            if not self.mass.music.library_supported(provider, MediaType.ARTIST):
1021                continue
1022            if not provider.is_streaming_provider:
1023                # matching on unique providers is pointless as they push (all) their content to MA
1024                continue
1025            if match := await self.match_provider(db_artist, provider):
1026                # 100% match, we update the db with the additional provider mapping(s)
1027                await self.add_provider_mappings(db_artist.item_id, match)
1028                cur_provider_domains.add(provider.domain)
1029
1030    def artist_from_item_mapping(self, item: ItemMapping) -> Artist:
1031        """Create an Artist object from an ItemMapping object."""
1032        domain, instance_id = None, None
1033        if prov := self.mass.get_provider(item.provider):
1034            domain = prov.domain
1035            instance_id = prov.instance_id
1036        return Artist.from_dict(
1037            {
1038                **item.to_dict(),
1039                "provider_mappings": [
1040                    {
1041                        "item_id": item.item_id,
1042                        "provider_domain": domain,
1043                        "provider_instance": instance_id,
1044                        "available": item.available,
1045                    }
1046                ],
1047            }
1048        )
1049
1050    def _validate_provider_filter(
1051        self, provider_instance_id_or_domain: str, provider_filter: str | None
1052    ) -> None:
1053        """Raise when a provider filter is set that does not match the requested provider."""
1054        if provider_filter is not None and provider_filter != provider_instance_id_or_domain:
1055            raise MusicAssistantError(
1056                f"provider_filter '{provider_filter}' does not match the requested "
1057                f"provider '{provider_instance_id_or_domain}'"
1058            )
1059
1060    async def _add_library_item(
1061        self, item: Artist | ItemMapping, overwrite_existing: bool = False
1062    ) -> int:
1063        """Add a new item record to the database."""
1064        # If item is an ItemMapping, convert it
1065        if isinstance(item, ItemMapping):
1066            item = self.artist_from_item_mapping(item)
1067        # enforce various artists name + id
1068        if compare_strings(item.name, VARIOUS_ARTISTS_NAME):
1069            item.mbid = VARIOUS_ARTISTS_MBID
1070        if item.mbid == VARIOUS_ARTISTS_MBID:
1071            item.name = VARIOUS_ARTISTS_NAME
1072        # no existing item matched: insert item
1073        db_id = await self.mass.music.database.insert(
1074            self.db_table,
1075            {
1076                "name": item.name,
1077                "sort_name": item.sort_name,
1078                "favorite": item.favorite,
1079                "metadata": serialize_to_json(item.metadata),
1080                "search_name": create_safe_string(item.name, True, True),
1081                "search_sort_name": create_safe_string(item.sort_name or "", True, True),
1082                "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
1083                "artist_type": item.artist_type,
1084            },
1085        )
1086        # update/set external id lookup table
1087        await self.set_external_ids(db_id, item.external_ids)
1088        # update/set provider_mappings table
1089        await self.set_provider_mappings(db_id, item.provider_mappings)
1090        self.logger.debug("added %s to database (id: %s)", item.name, db_id)
1091        return db_id
1092
1093    async def _update_library_item(
1094        self, item_id: str | int, update: Artist | ItemMapping, overwrite: bool = False
1095    ) -> None:
1096        """Update existing record in the database."""
1097        db_id = int(item_id)  # ensure integer
1098        cur_item = await self.get_library_item(db_id)
1099        if isinstance(update, ItemMapping):
1100            # NOTE that artist is the only mediatype where its accepted we
1101            # receive an itemmapping from streaming providers
1102            update = self.artist_from_item_mapping(update)
1103            metadata = cur_item.metadata
1104        else:
1105            metadata = update.metadata if overwrite else cur_item.metadata.update(update.metadata)
1106        cur_item.external_ids.update(update.external_ids)
1107        # enforce various artists name + id
1108        mbid = cur_item.mbid
1109        if (not mbid or overwrite) and getattr(update, "mbid", None):
1110            if compare_strings(update.name, VARIOUS_ARTISTS_NAME):
1111                update.mbid = VARIOUS_ARTISTS_MBID
1112            if update.mbid == VARIOUS_ARTISTS_MBID:
1113                update.name = VARIOUS_ARTISTS_NAME
1114
1115        name = update.name if overwrite else cur_item.name
1116        sort_name = update.sort_name if overwrite else cur_item.sort_name or update.sort_name
1117        await self.mass.music.database.update(
1118            self.db_table,
1119            {"item_id": db_id},
1120            {
1121                "name": name,
1122                "sort_name": sort_name,
1123                "metadata": serialize_to_json(metadata),
1124                "search_name": create_safe_string(name, True, True),
1125                "search_sort_name": create_safe_string(sort_name or "", True, True),
1126                "timestamp_added": int(update.date_added.timestamp())
1127                if update.date_added
1128                else UNSET,
1129                "artist_type": update.artist_type,
1130            },
1131        )
1132        self.logger.debug("updated %s in database: %s", update.name, db_id)
1133        # update/set external id lookup table
1134        await self.set_external_ids(
1135            db_id, update.external_ids if overwrite else cur_item.external_ids
1136        )
1137        # update/set provider_mappings table
1138        provider_mappings = (
1139            update.provider_mappings
1140            if overwrite
1141            else {*update.provider_mappings, *cur_item.provider_mappings}
1142        )
1143        await self.set_provider_mappings(db_id, provider_mappings, overwrite)
1144        self.logger.debug("updated %s in database: (id %s)", update.name, db_id)
1145
1146    async def _remove_music_artist_from_library(self, db_id: int, recursive: bool) -> None:
1147        # recursively also remove artist albums
1148        for db_row in await self.mass.music.database.get_rows_from_query(
1149            f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = :artist_id",
1150            {"artist_id": db_id},
1151            limit=5000,
1152        ):
1153            if not recursive:
1154                raise MusicAssistantError("Artist still has albums linked")
1155            with contextlib.suppress(MediaNotFoundError):
1156                await self.mass.music.albums.remove_item_from_library(db_row["album_id"])
1157        # recursively also remove artist tracks
1158        for db_row in await self.mass.music.database.get_rows_from_query(
1159            f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = :artist_id",
1160            {"artist_id": db_id},
1161            limit=5000,
1162        ):
1163            if not recursive:
1164                raise MusicAssistantError("Artist still has tracks linked")
1165            with contextlib.suppress(MediaNotFoundError):
1166                await self.mass.music.tracks.remove_item_from_library(db_row["track_id"])
1167
1168    async def _remove_author_narrator_from_library(self, db_id: int, recursive: bool) -> None:
1169        # recursively also remove author/ narrator audiobooks
1170        for db_row in await self.mass.music.database.get_rows_from_query(
1171            f"SELECT audiobook_id FROM {DB_TABLE_AUDIOBOOK_ARTISTS} WHERE artist_id = :artist_id",
1172            {"artist_id": db_id},
1173            limit=5000,
1174        ):
1175            if not recursive:
1176                raise MusicAssistantError("Artist still has audiobooks linked")
1177            with contextlib.suppress(MediaNotFoundError):
1178                await self.mass.music.audiobooks.remove_item_from_library(db_row["audiobook_id"])
1179
1180    async def _get_db_author_narrator_audiobooks(
1181        self, item_id: str, provider_instance_id_or_domain: str, artist_type: ArtistType
1182    ) -> list[Audiobook]:
1183        if db_author_narrator := await self.mass.music.artists.get_library_item_by_prov_id(
1184            item_id,
1185            provider_instance_id_or_domain,
1186        ):
1187            if db_author_narrator.artist_type != artist_type:
1188                self.logger.debug("Artist type must be %s.", artist_type)
1189                return []
1190            db_artist_id = int(db_author_narrator.item_id)  # ensure integer
1191            subquery = f"SELECT audiobook_id FROM {DB_TABLE_AUDIOBOOK_ARTISTS} WHERE artist_id = :artist_id"
1192            query = f"audiobooks.item_id in ({subquery})"
1193            return await self.mass.music.audiobooks.get_library_items_by_query(
1194                extra_query_parts=[query],
1195                extra_query_params={"artist_id": db_artist_id},
1196                provider_filter=[provider_instance_id_or_domain],
1197            )
1198        return []
1199
1200    def _parse_summary_row(self, db_row: Mapping[str, Any]) -> ArtistSummary:
1201        """Parse a raw summary db row into an ArtistSummary object."""
1202        item = cast("ArtistSummary", super()._parse_summary_row(db_row))
1203        item.artist_type = ArtistType(db_row["artist_type"])
1204        return item
1205