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