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