music-assistant-server

1"""Manage MediaItems of type Album."""
2
3from __future__ import annotations
4
5import contextlib
6from collections.abc import Iterable
7from typing import TYPE_CHECKING, Any, cast
8
9from music_assistant_models.auth import Scope
10from music_assistant_models.enums import AlbumType, ExternalID, MediaType, ProviderFeature
11from music_assistant_models.errors import InvalidDataError, MediaNotFoundError, MusicAssistantError
12from music_assistant_models.helpers import create_safe_string
13from music_assistant_models.media_items import (
14    Album,
15    AlbumSummary,
16    Artist,
17    ItemMapping,
18    MediaItemImage,
19    ProviderMapping,
20    Track,
21    UniqueList,
22)
23
24from music_assistant.constants import DB_TABLE_ALBUM_ARTISTS, DB_TABLE_ALBUM_TRACKS, DB_TABLE_ALBUMS
25from music_assistant.controllers.music.helpers import search_name_match_clause
26from music_assistant.helpers.compare import (
27    compare_album,
28    compare_artists,
29    compare_media_item,
30    loose_compare_strings,
31)
32from music_assistant.helpers.database import UNSET
33from music_assistant.helpers.json import serialize_to_json
34from music_assistant.models.music_provider import MusicProvider
35
36from .base import MediaControllerBase
37
38if TYPE_CHECKING:
39    from collections.abc import Mapping
40
41    from music_assistant import MusicAssistant
42
43
44class AlbumsController(MediaControllerBase[Album]):
45    """Controller managing MediaItems of type Album."""
46
47    db_table = DB_TABLE_ALBUMS
48    media_type = MediaType.ALBUM
49    item_cls = Album
50    summary_item_cls = AlbumSummary
51
52    def __init__(self, mass: MusicAssistant) -> None:
53        """Initialize class."""
54        super().__init__(mass)
55        # register (extra) api handlers
56        api_base = self.api_base
57        self.mass.register_api_command(
58            f"music/{api_base}/album_tracks", self.tracks, required_scope=Scope.LIBRARY_READ
59        )
60        self.mass.register_api_command(
61            f"music/{api_base}/album_versions", self.versions, required_scope=Scope.LIBRARY_READ
62        )
63
64    @property
65    def base_query(self) -> tuple[str, dict[str, Any]]:
66        """Return the base SELECT query for albums and its bound query params."""
67        query = f"""
68        SELECT
69            albums.*,
70            {self._external_ids_query()} AS external_ids,
71            {self._provider_mappings_query()} AS provider_mappings,
72            (SELECT JSON_GROUP_ARRAY(
73                json_object(
74                'item_id', artists.item_id,
75                'provider', 'library',
76                    'name', artists.name,
77                    'sort_name', artists.sort_name,
78                    'media_type', 'artist'
79                )) FROM artists JOIN album_artists on album_artists.album_id = albums.item_id  WHERE artists.item_id = album_artists.artist_id) AS artists
80            FROM albums"""
81        return query, {}
82
83    @property
84    def summary_query(self) -> tuple[str, dict[str, Any]]:
85        """Return the slim SELECT query used for album summary listings."""
86        artists_query = self._artist_mappings_summary_query(DB_TABLE_ALBUM_ARTISTS, "album_id")
87        query = f"""
88        SELECT
89            {self._summary_base_columns()},
90            albums.version,
91            albums.year,
92            albums.album_type,
93            {self._provider_mappings_query()} AS provider_mappings,
94            {artists_query} AS artists
95            FROM albums"""
96        return query, {}
97
98    async def get(
99        self,
100        item_id: str,
101        provider_instance_id_or_domain: str,
102        allow_update_metadata: bool = True,
103        recursive: bool = True,
104    ) -> Album:
105        """Return (full) details for a single media item."""
106        album = await super().get(
107            item_id,
108            provider_instance_id_or_domain,
109            allow_update_metadata=allow_update_metadata,
110        )
111        if not recursive:
112            return album
113
114        # append artist details to full album item (resolve ItemMappings)
115        album_artists: UniqueList[Artist | ItemMapping] = UniqueList()
116        for artist in album.artists:
117            if not isinstance(artist, ItemMapping):
118                album_artists.append(artist)
119                continue
120            with contextlib.suppress(MediaNotFoundError):
121                album_artists.append(
122                    await self.mass.music.artists.get(
123                        artist.item_id, artist.provider, allow_update_metadata=False
124                    )
125                )
126        album.artists = album_artists
127        return album
128
129    async def library_items(  # noqa: PLR0913
130        self,
131        favorite: bool | None = None,
132        search: str | None = None,
133        limit: int = 500,
134        offset: int = 0,
135        order_by: str = "sort_name",
136        provider: str | list[str] | None = None,
137        genre: int | list[int] | None = None,
138        played_only: bool = False,
139        album_types: list[AlbumType] | None = None,
140        *,
141        summary: bool = True,
142        reachable_via: list[str] | None = None,
143        **kwargs: Any,
144    ) -> list[Album]:
145        """
146        Get in-database albums.
147
148        :param favorite: Filter by favorite status.
149        :param search: Filter by search query.
150        :param limit: Maximum number of items to return.
151        :param offset: Number of items to skip.
152        :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
153        :param provider: Filter by provider instance ID (single string or list).
154        :param album_types: Filter by album types.
155        :param genre: Filter by genre id(s).
156        :param summary: When True (default), return slim summary items containing only the
157            fields needed for a list view. Set to False to get fully hydrated items.
158        :param reachable_via: Restrict results to items with a provider mapping reachable
159            through one of these provider instance ids (OR semantics). See
160            `MediaControllerBase.library_items` for the full semantics.
161        """
162        reachable_via = self._resolve_reachable_via(reachable_via)
163        if reachable_via is not None and not reachable_via:
164            return []
165        extra_query_params: dict[str, Any] = {}
166        extra_query_parts: list[str] = []
167        extra_join_parts: list[str] = []
168        artist_table_joined = False
169        # optional album type filter
170        if album_types:
171            extra_query_parts.append("albums.album_type IN :album_types")
172            extra_query_params["album_types"] = [x.value for x in album_types]
173        if order_by and "album_artist_name" in order_by:
174            # join artist table to allow sorting on artist name
175            extra_join_parts.append(
176                "JOIN album_artists ON album_artists.album_id = albums.item_id "
177                "JOIN artists ON artists.item_id = album_artists.artist_id "
178            )
179            artist_table_joined = True
180        if search and " - " in search:
181            # handle combined artist + title search
182            artist_str, title_str = search.split(" - ", 1)
183            search = None
184            title_str = create_safe_string(title_str, True, True)
185            artist_str = create_safe_string(artist_str, True, True)
186            extra_query_parts.append(
187                search_name_match_clause("albums", title_str, "search_title", extra_query_params)
188            )
189            artist_clause = "AND " + search_name_match_clause(
190                "artists", artist_str, "search_artist", extra_query_params
191            )
192            # use join with artists table to filter on artist name
193            extra_join_parts.append(
194                "JOIN album_artists ON album_artists.album_id = albums.item_id "
195                "JOIN artists ON artists.item_id = album_artists.artist_id " + artist_clause
196                if not artist_table_joined
197                else artist_clause
198            )
199            artist_table_joined = True
200        result = await self.get_library_items_by_query(
201            favorite=favorite,
202            search=search,
203            genre_ids=genre,
204            limit=limit,
205            offset=offset,
206            order_by=order_by,
207            provider_filter=self._provider_filter_considering_reachability(provider, reachable_via),
208            extra_query_parts=extra_query_parts,
209            extra_query_params=extra_query_params,
210            extra_join_parts=extra_join_parts,
211            played_only=played_only,
212            in_library_only=True,
213            summary=summary,
214            reachable_via=reachable_via,
215        )
216
217        # Calculate how many more items we need to reach the original limit
218        remaining_limit = limit - len(result)
219
220        if search and len(result) < 25 and not offset and remaining_limit > 0:
221            # append artist items to result
222            search = create_safe_string(search, True, True)
223            artist_clause = "AND " + search_name_match_clause(
224                "artists", search, "search_artist", extra_query_params
225            )
226            extra_join_parts.append(
227                "JOIN album_artists ON album_artists.album_id = albums.item_id "
228                "JOIN artists ON artists.item_id = album_artists.artist_id " + artist_clause
229                if not artist_table_joined
230                else artist_clause
231            )
232            existing_uris = {item.uri for item in result}
233
234            for album in await self.get_library_items_by_query(
235                favorite=favorite,
236                search=None,
237                limit=remaining_limit,
238                order_by=order_by,
239                provider_filter=self._provider_filter_considering_reachability(
240                    provider, reachable_via
241                ),
242                extra_query_parts=extra_query_parts,
243                extra_query_params=extra_query_params,
244                extra_join_parts=extra_join_parts,
245                in_library_only=True,
246                summary=summary,
247                reachable_via=reachable_via,
248            ):
249                # prevent duplicates (when artist is also in the title)
250                if album.uri not in existing_uris:
251                    result.append(album)
252                    # Stop if we've reached the original limit
253                    if len(result) >= limit:
254                        break
255        return result
256
257    async def library_count(
258        self, favorite_only: bool = False, album_types: list[AlbumType] | None = None
259    ) -> int:
260        """
261        Return the number of albums in the library.
262
263        Restricted to the providers the current user is allowed to see when that user
264        has a provider filter set.
265
266        :param favorite_only: Only count albums marked as favorite.
267        :param album_types: Only count albums of these types.
268        """
269        sql_query = f"SELECT item_id FROM {self.db_table}"
270        query_parts: list[str] = []
271        query_params: dict[str, Any] = {}
272        if favorite_only:
273            query_parts.append("favorite = 1")
274        if album_types:
275            query_parts.append("albums.album_type IN :album_types")
276            query_params["album_types"] = [x.value for x in album_types]
277        if provider_filter := self._ensure_provider_filter(None):
278            query_parts.append(
279                self._provider_filter_clause(query_params, provider_filter, in_library_only=True)
280            )
281        if query_parts:
282            sql_query += f" WHERE {' AND '.join(query_parts)}"
283        return await self.mass.music.database.get_count_from_query(sql_query, query_params)
284
285    async def remove_item_from_library(self, item_id: str | int, recursive: bool = True) -> None:
286        """Delete item from the library(database)."""
287        db_id = int(item_id)  # ensure integer
288        # recursively also remove album tracks
289        for db_track in await self.get_library_album_tracks(db_id):
290            if not recursive:
291                raise MusicAssistantError("Album still has tracks linked")
292            with contextlib.suppress(MediaNotFoundError):
293                await self.mass.music.tracks.remove_item_from_library(db_track.item_id)
294        # delete entry(s) from albumtracks table
295        await self.mass.music.database.delete(DB_TABLE_ALBUM_TRACKS, {"album_id": db_id})
296        # delete entry(s) from album artists table
297        await self.mass.music.database.delete(DB_TABLE_ALBUM_ARTISTS, {"album_id": db_id})
298        # delete the album itself from db
299        # this will raise if the item still has references and recursive is false
300        await super().remove_item_from_library(item_id)
301
302    async def set_release_group(
303        self,
304        album_item_id: int,
305        release_group_mbid: str,
306    ) -> None:
307        """
308        Persist a MusicBrainz release-group ID on a library album, idempotently.
309
310        :param album_item_id: Library album item_id (database id).
311        :param release_group_mbid: MusicBrainz release-group UUID to set.
312        """
313        if not release_group_mbid:
314            return
315        try:
316            album = await self.get_library_item(album_item_id)
317        except MusicAssistantError as err:
318            self.logger.debug("set_release_group: cannot load album %s: %s", album_item_id, err)
319            return
320        # Refuse to overwrite — keeps tag-sourced or already-enriched IDs authoritative.
321        if album.get_external_id(ExternalID.MB_RELEASEGROUP):
322            self.logger.debug(
323                "set_release_group: album %s already has MB_RELEASEGROUP — keeping",
324                album_item_id,
325            )
326            return
327        album.add_external_id(ExternalID.MB_RELEASEGROUP, release_group_mbid)
328        await self.update_item_in_library(album_item_id, album)
329        self.logger.debug(
330            "set_release_group: wrote %s onto album %s", release_group_mbid, album_item_id
331        )
332
333    async def tracks(
334        self,
335        item_id: str,
336        provider_instance_id_or_domain: str,
337        in_library_only: bool = False,
338    ) -> list[Track]:
339        """Return album tracks for the given provider album id."""
340        # always check if we have a library item for this album
341        library_album = await self.get_library_item_by_prov_id(
342            item_id, provider_instance_id_or_domain
343        )
344        if not library_album:
345            album_tracks = await self._get_provider_album_tracks(
346                item_id, provider_instance_id_or_domain
347            )
348            # some album-track listings omit the parent album and its image; backfill both
349            # from the provider album so the queue shows the album name and artwork.
350            if album_tracks and (not album_tracks[0].album or not album_tracks[0].image):
351                prov_album = await self.get_provider_item(item_id, provider_instance_id_or_domain)
352                album_mapping = ItemMapping.from_item(prov_album)
353                for track in album_tracks:
354                    if prov_album.image and not track.image:
355                        track.metadata.add_image(prov_album.image)
356                    if track.album is None:
357                        track.album = album_mapping
358            return album_tracks
359
360        # respect the current user's provider filter (if any) for both the
361        # in-library tracks and the live provider fetches below
362        allowed_providers = self._ensure_provider_filter(None)
363        db_items = await self.get_library_album_tracks(
364            library_album.item_id, provider_filter=allowed_providers
365        )
366        result: list[Track] = list(db_items)
367        if in_library_only:
368            # return in-library items only
369            return sorted(db_items, key=lambda x: (x.disc_number, x.track_number))
370
371        # return all (unique) items from all providers
372        # because we are returning the items from all providers combined,
373        # we need to make sure that we don't return duplicates
374        unique_ids: set[str] = {f"{x.disc_number}.{x.track_number}" for x in db_items}
375        unique_ids.update({f"{x.name.lower()}.{x.version.lower()}" for x in db_items})
376        for db_item in db_items:
377            unique_ids.update(x.item_id for x in db_item.provider_mappings)
378        for provider_mapping in library_album.provider_mappings:
379            if (
380                allowed_providers is not None
381                and provider_mapping.provider_instance not in allowed_providers
382            ):
383                continue
384            provider_tracks = await self._get_provider_album_tracks(
385                provider_mapping.item_id, provider_mapping.provider_instance
386            )
387            for provider_track in provider_tracks:
388                # In some cases (looking at you YTM) the disc/track number is not obtained from
389                # library_tracks. Ensure to update the disc/track number when interacting with
390                # album tracks
391                db_track = next(
392                    (
393                        x
394                        for x in db_items
395                        if x.sort_name == provider_track.sort_name
396                        and x.version == provider_track.version
397                    ),
398                    None,
399                )
400                if (
401                    db_track
402                    and db_track.track_number == 0
403                    and db_track.track_number != provider_track.track_number
404                ):
405                    await self._set_album_track(
406                        db_id=int(library_album.item_id),
407                        db_track_id=int(db_track.item_id),
408                        track=provider_track,
409                    )
410                if provider_track.item_id in unique_ids:
411                    continue
412                unique_id = f"{provider_track.disc_number}.{provider_track.track_number}"
413                if unique_id in unique_ids:
414                    continue
415                unique_id = f"{provider_track.name.lower()}.{provider_track.version.lower()}"
416                if unique_id in unique_ids:
417                    continue
418                unique_ids.add(unique_id)
419                provider_track.album = library_album
420                # always prefer album image
421                album_images = [library_album.image] if library_album.image else []
422                track_images: list[MediaItemImage] = provider_track.metadata.images or []
423                provider_track.metadata.images = UniqueList(album_images + track_images)
424                result.append(provider_track)
425        # NOTE: we need to return the results sorted on disc/track here
426        # to ensure the correct order at playback
427        return sorted(result, key=lambda x: (x.disc_number, x.track_number))
428
429    async def versions(
430        self,
431        item_id: str,
432        provider_instance_id_or_domain: str,
433    ) -> UniqueList[Album]:
434        """Return all versions of an album we can find on all providers."""
435        album = await self.get_provider_item(item_id, provider_instance_id_or_domain)
436        streaming_search_query = (
437            f"{album.artists[0].name} - {album.name}" if album.artists else album.name
438        )
439        result: UniqueList[Album] = UniqueList()
440        for provider_id in self.mass.music.get_unique_providers():
441            provider = self.mass.get_provider(provider_id)
442            if not provider or not isinstance(provider, MusicProvider):
443                continue
444            if not self.mass.music.library_supported(provider, MediaType.ALBUM):
445                continue
446            # TODO: filter by artists in db for non-streaming providers
447            search_query = streaming_search_query if provider.is_streaming_provider else album.name
448            result.extend(
449                prov_item
450                for prov_item in await self.search(search_query, provider_id)
451                if loose_compare_strings(album.name, prov_item.name)
452                and compare_artists(prov_item.artists, album.artists, any_match=True)
453                # make sure that the 'base' version is NOT included
454                and not album.provider_mappings.intersection(prov_item.provider_mappings)
455            )
456        return result
457
458    async def get_library_album_tracks(
459        self,
460        item_id: str | int,
461        provider_filter: list[str] | None = None,
462    ) -> list[Track]:
463        """
464        Return in-database album tracks for the given database album.
465
466        :param item_id: The library item ID of the album.
467        :param provider_filter: Optional provider instance ID(s) to limit the result to.
468        """
469        db_id = int(item_id)  # ensure integer
470        # pass the album id as preferred album so the track_album subquery in the
471        # base query returns this album's disc/track numbers for tracks that
472        # appear on multiple albums
473        return await self.mass.music.tracks.get_library_items_by_query(
474            provider_filter=provider_filter,
475            extra_query_parts=[
476                f"tracks.item_id IN (SELECT track_id FROM {DB_TABLE_ALBUM_TRACKS} "
477                "WHERE album_id = :album_id)"
478            ],
479            extra_query_params={"album_id": db_id, "preferred_album_id": db_id},
480        )
481
482    async def add_item_mapping_as_album_to_library(self, item: ItemMapping) -> Album:
483        """
484        Add an ItemMapping as an Album to the library.
485
486        This is only used in special occasions as is basically adds an album
487        to the db without a lot of mandatory data, such as artists.
488        """
489        album = self.album_from_item_mapping(item)
490        return await self.add_item_to_library(album)
491
492    async def match_provider(
493        self, db_album: Album, provider: MusicProvider, strict: bool = True
494    ) -> list[ProviderMapping]:
495        """
496        Try to find match on (streaming) provider for the provided (database) album.
497
498        This is used to link objects of different providers/qualities together.
499        """
500        self.logger.debug("Trying to match album %s on provider %s", db_album.name, provider.name)
501        matches: list[ProviderMapping] = []
502        artist_name = db_album.artists[0].name
503        search_str = f"{artist_name} - {db_album.name}"
504        search_result = await self.search(search_str, provider.instance_id)
505        for search_result_item in search_result:
506            if not search_result_item.available:
507                continue
508            if not compare_media_item(db_album, search_result_item, strict=strict):
509                continue
510            # we must fetch the full album version, search results can be simplified objects
511            prov_album = await self.get_provider_item(
512                search_result_item.item_id,
513                search_result_item.provider,
514                fallback=search_result_item,
515            )
516            if compare_album(db_album, prov_album, strict=strict):
517                # 100% match
518                matches.extend(prov_album.provider_mappings)
519        if not matches:
520            self.logger.debug(
521                "Could not find match for Album %s on provider %s",
522                db_album.name,
523                provider.name,
524            )
525        return matches
526
527    async def match_providers(self, db_album: Album) -> None:
528        """
529        Try to find match on all (streaming) providers for the provided (database) album.
530
531        This is used to link objects of different providers/qualities together.
532        """
533        if db_album.provider != "library":
534            return  # Matching only supported for database items
535        if not db_album.artists:
536            return  # guard
537
538        # try to find match on all providers
539        processed_domains = set()
540        for provider in self.mass.music.providers:
541            if provider.domain in processed_domains:
542                continue
543            if ProviderFeature.SEARCH not in provider.supported_features:
544                continue
545            if not self.mass.music.library_supported(provider, MediaType.ALBUM):
546                continue
547            if not provider.is_streaming_provider:
548                # matching on unique providers is pointless as they push (all) their content to MA
549                continue
550            if match := await self.match_provider(db_album, provider):
551                # 100% match, we update the db with the additional provider mapping(s)
552                await self.add_provider_mappings(db_album.item_id, match)
553                processed_domains.add(provider.domain)
554
555    def album_from_item_mapping(self, item: ItemMapping) -> Album:
556        """Create an Album object from an ItemMapping object."""
557        domain, instance_id = None, None
558        if prov := self.mass.get_provider(item.provider):
559            domain = prov.domain
560            instance_id = prov.instance_id
561        return Album.from_dict(
562            {
563                **item.to_dict(),
564                "provider_mappings": [
565                    {
566                        "item_id": item.item_id,
567                        "provider_domain": domain,
568                        "provider_instance": instance_id,
569                        "available": item.available,
570                    }
571                ],
572            }
573        )
574
575    async def _add_library_item(self, item: Album, overwrite_existing: bool = False) -> int:
576        """Add a new record to the database."""
577        if not isinstance(item, Album):  # TODO: Remove this once the codebase is fully typed
578            msg = "Not a valid Album object (ItemMapping can not be added to db)"  # type: ignore[unreachable]
579            raise InvalidDataError(msg)
580        db_id = await self.mass.music.database.insert(
581            self.db_table,
582            {
583                "name": item.name,
584                "sort_name": item.sort_name,
585                "version": item.version,
586                "favorite": item.favorite,
587                "album_type": item.album_type,
588                "year": item.year,
589                "metadata": serialize_to_json(item.metadata),
590                "search_name": create_safe_string(item.name, True, True),
591                "search_sort_name": create_safe_string(item.sort_name or "", True, True),
592                "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
593            },
594        )
595        # update/set external id lookup table
596        await self.set_external_ids(db_id, item.external_ids)
597        # update/set provider_mappings table
598        await self.set_provider_mappings(db_id, item.provider_mappings)
599        # set track artist(s)
600        await self._set_album_artists(db_id, item.artists)
601        self.logger.debug("added %s to database (id: %s)", item.name, db_id)
602        return db_id
603
604    async def _update_library_item(
605        self, item_id: str | int, update: Album, overwrite: bool = False
606    ) -> None:
607        """Update existing record in the database."""
608        db_id = int(item_id)  # ensure integer
609        cur_item = await self.get_library_item(db_id)
610        metadata = update.metadata if overwrite else cur_item.metadata.update(update.metadata)
611        if getattr(update, "album_type", AlbumType.UNKNOWN) != AlbumType.UNKNOWN:
612            album_type = update.album_type
613        else:
614            album_type = cur_item.album_type
615        cur_item.external_ids.update(update.external_ids)
616        name = update.name if overwrite else cur_item.name
617        sort_name = update.sort_name if overwrite else cur_item.sort_name or update.sort_name
618        await self.mass.music.database.update(
619            self.db_table,
620            {"item_id": db_id},
621            {
622                "name": name,
623                "sort_name": sort_name,
624                "version": update.version if overwrite else cur_item.version or update.version,
625                "year": update.year if overwrite else cur_item.year or update.year,
626                "album_type": album_type.value,
627                "metadata": serialize_to_json(metadata),
628                "search_name": create_safe_string(name, True, True),
629                "search_sort_name": create_safe_string(sort_name or "", True, True),
630                "timestamp_added": int(update.date_added.timestamp())
631                if update.date_added
632                else UNSET,
633            },
634        )
635        # update/set external id lookup table
636        await self.set_external_ids(
637            db_id, update.external_ids if overwrite else cur_item.external_ids
638        )
639        # update/set provider_mappings table
640        provider_mappings = (
641            update.provider_mappings
642            if overwrite
643            else {*update.provider_mappings, *cur_item.provider_mappings}
644        )
645        await self.set_provider_mappings(db_id, provider_mappings, overwrite)
646        # set album artist(s)
647        artists = update.artists if overwrite else cur_item.artists + update.artists
648        await self._set_album_artists(db_id, artists, overwrite=overwrite)
649        self.logger.debug("updated %s in database: (id %s)", update.name, db_id)
650
651    async def _get_provider_album_tracks(
652        self, item_id: str, provider_instance_id_or_domain: str
653    ) -> list[Track]:
654        """Return album tracks for the given provider album id."""
655        if prov := self.mass.get_provider(provider_instance_id_or_domain):
656            prov = cast("MusicProvider", prov)
657            return await prov.get_album_tracks(item_id)
658        return []
659
660    async def _set_album_artists(
661        self,
662        db_id: int,
663        artists: Iterable[Artist | ItemMapping],
664        overwrite: bool = False,
665    ) -> None:
666        """Store Album Artists."""
667        if overwrite:
668            # on overwrite, clear the album_artists table first
669            await self.mass.music.database.delete(
670                DB_TABLE_ALBUM_ARTISTS,
671                {
672                    "album_id": db_id,
673                },
674            )
675        for artist in artists:
676            await self._set_album_artist(db_id, artist=artist, overwrite=overwrite)
677
678    async def _set_album_artist(
679        self, db_id: int, artist: Artist | ItemMapping, overwrite: bool = False
680    ) -> ItemMapping:
681        """Store Album Artist info."""
682        db_artist: Artist | ItemMapping | None = None
683        if artist.provider == "library":
684            db_artist = artist
685        elif existing := await self.mass.music.artists.get_library_item_by_prov_id(
686            artist.item_id, artist.provider
687        ):
688            db_artist = existing
689
690        if not db_artist or overwrite:
691            # Convert ItemMapping to Artist if needed
692            artist_to_add = (
693                self.mass.music.artists.artist_from_item_mapping(artist)
694                if isinstance(artist, ItemMapping)
695                else artist
696            )
697            db_artist = await self.mass.music.artists.add_item_to_library(
698                artist_to_add, overwrite_existing=overwrite
699            )
700        # write (or update) record in album_artists table
701        await self.mass.music.database.insert_or_replace(
702            DB_TABLE_ALBUM_ARTISTS,
703            {
704                "album_id": db_id,
705                "artist_id": int(db_artist.item_id),
706            },
707        )
708        return ItemMapping.from_item(db_artist)
709
710    async def _set_album_track(self, db_id: int, db_track_id: int, track: Track) -> None:
711        """Store Album Track info."""
712        # write (or update) record in album_tracks table
713        await self.mass.music.database.insert_or_replace(
714            DB_TABLE_ALBUM_TRACKS,
715            {
716                "album_id": db_id,
717                "track_id": db_track_id,
718                "track_number": track.track_number,
719                "disc_number": track.disc_number,
720            },
721        )
722
723    def _parse_summary_row(self, db_row: Mapping[str, Any]) -> AlbumSummary:
724        """Parse a raw summary db row into an AlbumSummary object."""
725        item = cast("AlbumSummary", super()._parse_summary_row(db_row))
726        item.version = db_row["version"] or ""
727        item.year = db_row["year"]
728        item.album_type = AlbumType(db_row["album_type"])
729        item.artists = self._parse_summary_artist_mappings(db_row)
730        return item
731