music-assistant-server

1"""Base (ABC) MediaType specific controller."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from abc import ABCMeta, abstractmethod
8from collections.abc import Iterable
9from contextlib import suppress
10from contextvars import ContextVar
11from dataclasses import dataclass
12from datetime import UTC, datetime
13from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast, final, overload
14
15from music_assistant_models.auth import Scope
16from music_assistant_models.enums import (
17    EventType,
18    ExternalID,
19    ImageType,
20    MediaType,
21    ProviderFeature,
22    ProviderType,
23)
24from music_assistant_models.errors import (
25    InsufficientPermissions,
26    MediaNotFoundError,
27    ProviderUnavailableError,
28)
29from music_assistant_models.helpers import create_safe_string, get_global_cache_value
30from music_assistant_models.media_items import (
31    AudioFormat,
32    ItemMapping,
33    ItemMappingSummary,
34    MediaCollection,
35    MediaItemImage,
36    MediaItemMetadata,
37    MediaItemMetadataSummary,
38    MediaItemSummaryType,
39    MediaItemType,
40    ProviderMapping,
41    UniqueList,
42)
43
44from music_assistant.constants import (
45    DB_TABLE_AUDIO_ANALYSIS,
46    DB_TABLE_EXTERNAL_ID_LOOKUP,
47    DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION,
48    DB_TABLE_GENRE_MEDIA_ITEM_MAPPING,
49    DB_TABLE_PLAYLOG,
50    DB_TABLE_PROVIDER_MAPPINGS,
51    MASS_LOGGER_NAME,
52)
53from music_assistant.controllers.music.helpers import search_name_match_clause
54from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
55from music_assistant.helpers.collections import (
56    get_collection_item_id,
57    get_collection_name_from_item_id,
58)
59from music_assistant.helpers.compare import compare_media_item
60from music_assistant.helpers.database import UNSET
61from music_assistant.helpers.json import json_loads, serialize_to_json
62from music_assistant.helpers.util import guard_single_request, parse_optional_bool
63
64if TYPE_CHECKING:
65    from collections.abc import AsyncGenerator, Mapping
66
67    from music_assistant import MusicAssistant
68    from music_assistant.models.music_provider import MusicProvider
69    from music_assistant.models.plugin import PluginProvider
70
71
72ItemCls = TypeVar("ItemCls", bound="MediaItemType")
73
74
75JSON_KEYS = (
76    "artists",
77    "track_album",
78    "metadata",
79    "provider_mappings",
80    "external_ids",
81    "narrators",
82    "authors",
83    "genre_aliases",
84    "supported_mediatypes",
85    "translation_params",
86    "audiobook_artists",
87)
88
89# When set (task-local), per-item MEDIA_ITEM_ADDED/UPDATED events and the on_item_updated
90# provider write-back are suppressed, so bulk operations (provider sync, provider cleanup)
91# don't flood subscribers with one event per touched item.
92SUPPRESS_MEDIA_ITEM_UPDATES: ContextVar[bool] = ContextVar(
93    "SUPPRESS_MEDIA_ITEM_UPDATES", default=False
94)
95
96SORT_KEYS = {
97    # sqlite has no builtin support for natural sorting
98    # so we have use an additional column for this
99    # this also improves searching and sorting performance
100    "name": "search_name ASC",
101    "name_desc": "search_name DESC",
102    "duration": "duration ASC",
103    "duration_desc": "duration DESC",
104    "sort_name": "search_sort_name ASC",
105    "sort_name_desc": "search_sort_name DESC",
106    "timestamp_added": "timestamp_added ASC",
107    "timestamp_added_desc": "timestamp_added DESC",
108    "timestamp_modified": "timestamp_modified ASC",
109    "timestamp_modified_desc": "timestamp_modified DESC",
110    "last_played": "last_played ASC",
111    "last_played_desc": "last_played DESC",
112    "play_count": "play_count ASC",
113    "play_count_desc": "play_count DESC",
114    "year": "year ASC",
115    "year_desc": "year DESC",
116    "position": "position ASC",
117    "position_desc": "position DESC",
118    "artist_name": "artists.search_name ASC, year DESC",
119    "artist_name_desc": "artists.search_name DESC, year DESC",
120    "random": "RANDOM()",
121    "random_play_count": "RANDOM(), play_count ASC",
122}
123
124
125@dataclass(slots=True)
126class LibraryItemSyncDetails:
127    """
128    Lightweight snapshot of a library item with just the fields the library sync needs.
129
130    Used by the provider sync loops to detect (un)changed items without hydrating
131    full MediaItem objects from the database.
132    """
133
134    item_id: int
135    favorite: bool
136    date_added: datetime
137    provider_mappings: set[ProviderMapping]
138
139
140@dataclass(slots=True)
141class TrackSyncDetails(LibraryItemSyncDetails):
142    """Lightweight sync snapshot of a library track."""
143
144    has_album: bool
145
146
147@dataclass(slots=True)
148class AudiobookSyncDetails(LibraryItemSyncDetails):
149    """Lightweight sync snapshot of a library audiobook."""
150
151    author_is_str: bool
152    narrator_is_str: bool
153    fully_played: bool | None
154    resume_position_ms: int | None
155
156
157class MediaControllerBase[ItemCls: "MediaItemType"](metaclass=ABCMeta):
158    """Base model for controller managing a MediaType."""
159
160    media_type: MediaType
161    item_cls: type[MediaItemType]
162    summary_item_cls: type[MediaItemSummaryType]
163    db_table: str
164
165    def __init__(self, mass: MusicAssistant) -> None:
166        """Initialize class."""
167        self.mass = mass
168        self.logger = logging.getLogger(f"{MASS_LOGGER_NAME}.music.{self.media_type.value}")
169        # register (base) api handlers
170        self.api_base = api_base = f"{self.media_type}s"
171        self.mass.register_api_command(
172            f"music/{api_base}/count", self.library_count, required_scope=Scope.LIBRARY_READ
173        )
174        self.mass.register_api_command(
175            f"music/{api_base}/library_items",
176            self.library_items,
177            required_scope=Scope.LIBRARY_READ,
178            allow_impersonation=True,
179        )
180        self.mass.register_api_command(
181            f"music/{api_base}/get", self.get, required_scope=Scope.LIBRARY_READ
182        )
183        self.mass.register_api_command(
184            f"music/{api_base}/get_by_external_id",
185            self.get_library_item_by_external_id,
186            required_scope=Scope.LIBRARY_READ,
187        )
188        self.mass.register_api_command(
189            f"music/{api_base}/get_collection",
190            self.get_collection,
191            required_scope=Scope.LIBRARY_READ,
192            allow_impersonation=True,
193        )
194        # Backward compatibility alias - prefer the generic "get" endpoint
195        self.mass.register_api_command(
196            f"music/{api_base}/get_{self.media_type}",
197            self.get,
198            required_scope=Scope.LIBRARY_READ,
199            alias=True,
200        )
201        self.mass.register_api_command(
202            f"music/{api_base}/update",
203            self.update_item_in_library,
204            required_scope=Scope.LIBRARY_MANAGE,
205        )
206        self.mass.register_api_command(
207            f"music/{api_base}/remove",
208            self.remove_item_from_library,
209            required_scope=Scope.LIBRARY_MANAGE,
210        )
211        self._db_add_lock = asyncio.Lock()
212
213    @property
214    def translation_owner(self) -> str:
215        """Return the "core.music" namespace these media controllers' translation strings live under."""
216        return "core.music"
217
218    @property
219    def base_query(self) -> tuple[str, dict[str, Any]]:
220        """
221        Return the base SELECT query for this media type and its bound query params.
222
223        Override in a subclass to customize the query (extra joins/columns) and/or to
224        inject dynamic, parameterized filters.
225        """
226        query = f"""
227        SELECT
228            {self.db_table}.*,
229            {self._external_ids_query()} AS external_ids,
230            {self._provider_mappings_query()} AS provider_mappings
231            FROM {self.db_table} """
232        return query, {}
233
234    @property
235    def summary_query(self) -> tuple[str, dict[str, Any]]:
236        """
237        Return the slim SELECT query used for summary listings and its bound query params.
238
239        Selects only the columns needed to build summary items. Override in a subclass
240        to select additional per-type columns.
241        """
242        query = f"""
243        SELECT
244            {self._summary_base_columns()},
245            {self._provider_mappings_query()} AS provider_mappings
246            FROM {self.db_table} """
247        return query, {}
248
249    @final
250    async def add_item_to_library(
251        self,
252        item: ItemCls,
253        overwrite_existing: bool = False,
254    ) -> ItemCls:
255        """Add item to library and return the new (or updated) database item."""
256        new_item = False
257        # batch the many writes of an item add/update into a single commit
258        async with self.mass.music.database.deferred_commit():
259            # check for existing item first
260            if library_id := await self._get_library_item_by_match(item):
261                # update existing item
262                await self._update_library_item(library_id, item, overwrite=overwrite_existing)
263            else:
264                # actually add a new item in the library db
265                self.mass.music.match_provider_instances(item)
266                async with self._db_add_lock:
267                    library_id = await self._add_library_item(item)
268                    new_item = True
269        # return final library_item
270        library_item = await self.get_library_item(library_id)
271        if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
272            self.mass.signal_event(
273                EventType.MEDIA_ITEM_ADDED if new_item else EventType.MEDIA_ITEM_UPDATED,
274                library_item.uri,
275                library_item,
276            )
277        return library_item
278
279    @final
280    async def update_item_in_library(
281        self, item_id: str | int, update: ItemCls, overwrite: bool = False
282    ) -> ItemCls:
283        """Update existing library record in the library database."""
284        self.mass.music.match_provider_instances(update)
285        # batch the many writes of an item update into a single commit
286        async with self.mass.music.database.deferred_commit():
287            await self._update_library_item(item_id, update, overwrite=overwrite)
288        # return the updated object
289        library_item = await self.get_library_item(item_id)
290        if SUPPRESS_MEDIA_ITEM_UPDATES.get():
291            # during a sync the update originates from the provider itself,
292            # so skip both the event and the write-back to that provider
293            return library_item
294        # drop cached artwork for the updated item so replaced art is served fresh
295        for img in library_item.metadata.images or []:
296            await self.mass.metadata.invalidate_image_cache(img.provider, img.path)
297        self.mass.signal_event(
298            EventType.MEDIA_ITEM_UPDATED,
299            library_item.uri,
300            library_item,
301        )
302        # notify music providers of the update so they can sync their own storage
303        for prov_mapping in library_item.provider_mappings:
304            if provider := self.mass.get_provider(prov_mapping.provider_instance):
305                if provider.type != ProviderType.MUSIC:
306                    continue
307                provider = cast("MusicProvider", provider)
308                await provider.on_item_updated(library_item)
309        return library_item
310
311    async def remove_item_from_library(self, item_id: str | int, recursive: bool = True) -> None:
312        """Delete library record from the database."""
313        db_id = int(item_id)  # ensure integer
314        library_item = await self.get_library_item(db_id)
315        assert library_item, f"Item does not exist: {db_id}"
316        # delete item
317        await self.mass.music.database.delete(
318            self.db_table,
319            {"item_id": db_id},
320        )
321        # update provider_mappings table
322        await self.mass.music.database.delete(
323            DB_TABLE_PROVIDER_MAPPINGS,
324            {"media_type": self.media_type.value, "item_id": db_id},
325        )
326        # cleanup external_id_lookup table
327        await self.mass.music.database.delete(
328            DB_TABLE_EXTERNAL_ID_LOOKUP,
329            {"media_type": self.media_type.value, "item_id": db_id},
330        )
331        # cleanup playlog table
332        await self.mass.music.database.delete(
333            DB_TABLE_PLAYLOG,
334            {
335                "media_type": self.media_type.value,
336                "item_id": db_id,
337                "provider": "library",
338            },
339        )
340        for prov_mapping in library_item.provider_mappings:
341            await self.mass.music.database.delete(
342                DB_TABLE_PLAYLOG,
343                {
344                    "media_type": self.media_type.value,
345                    "item_id": prov_mapping.item_id,
346                    "provider": prov_mapping.provider_instance,
347                },
348            )
349            # cleanup audio analysis rows for this provider mapping
350            for prov_key in (prov_mapping.provider_domain, prov_mapping.provider_instance):
351                await self.mass.music.database.delete(
352                    DB_TABLE_AUDIO_ANALYSIS,
353                    {
354                        "media_type": self.media_type.value,
355                        "item_id": prov_mapping.item_id,
356                        "provider": prov_key,
357                    },
358                )
359        # delete genre exclusions for this media item
360        await self.mass.music.database.delete(
361            DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION,
362            {"media_type": self.media_type.value, "media_id": db_id},
363        )
364        # NOTE: this does not delete any references to this item in other records,
365        # this is handled/overridden in the mediatype specific controllers
366        # drop cached artwork for the removed item
367        for img in library_item.metadata.images or []:
368            await self.mass.metadata.invalidate_image_cache(img.provider, img.path)
369        self.mass.signal_event(EventType.MEDIA_ITEM_DELETED, library_item.uri, library_item)
370        self.logger.debug("deleted item with id %s from database", db_id)
371
372    async def library_count(self, favorite_only: bool = False) -> int:
373        """
374        Return the number of items in the library.
375
376        Restricted to the providers the current user is allowed to see when that user
377        has a provider filter set.
378
379        :param favorite_only: Only count items marked as favorite.
380        """
381        query_parts: list[str] = []
382        query_params: dict[str, Any] = {}
383        if favorite_only:
384            query_parts.append("favorite = 1")
385        if provider_filter := self._ensure_provider_filter(None):
386            query_parts.append(
387                self._provider_filter_clause(query_params, provider_filter, in_library_only=True)
388            )
389        if not query_parts:
390            return await self.mass.music.database.get_count(self.db_table)
391        sql_query = f"SELECT item_id FROM {self.db_table} WHERE {' AND '.join(query_parts)}"
392        return await self.mass.music.database.get_count_from_query(sql_query, query_params)
393
394    if TYPE_CHECKING:
395
396        @overload
397        async def library_items(
398            self,
399            favorite: bool | None = None,
400            search: str | None = None,
401            limit: int = 500,
402            offset: int = 0,
403            order_by: str = "sort_name",
404            provider: str | list[str] | None = None,
405            genre: int | list[int] | None = None,
406            played_only: bool = False,
407            *,
408            summary: bool = True,
409            collapse_collections: Literal[False] = False,
410            **kwargs: Any,
411        ) -> list[ItemCls]: ...
412
413        @overload
414        async def library_items(
415            self,
416            favorite: bool | None = None,
417            search: str | None = None,
418            limit: int = 500,
419            offset: int = 0,
420            order_by: str = "sort_name",
421            provider: str | list[str] | None = None,
422            genre: int | list[int] | None = None,
423            played_only: bool = False,
424            *,
425            summary: bool = True,
426            collapse_collections: Literal[True],
427            **kwargs: Any,
428        ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]: ...
429
430        @overload
431        async def library_items(
432            self,
433            favorite: bool | None = None,
434            search: str | None = None,
435            limit: int = 500,
436            offset: int = 0,
437            order_by: str = "sort_name",
438            provider: str | list[str] | None = None,
439            genre: int | list[int] | None = None,
440            played_only: bool = False,
441            *,
442            summary: bool = True,
443            collapse_collections: bool,
444            **kwargs: Any,
445        ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]: ...
446
447    async def library_items(
448        self,
449        favorite: bool | None = None,
450        search: str | None = None,
451        limit: int = 500,
452        offset: int = 0,
453        order_by: str = "sort_name",
454        provider: str | list[str] | None = None,
455        genre: int | list[int] | None = None,
456        played_only: bool = False,
457        *,
458        summary: bool = True,
459        collapse_collections: bool = False,
460        **kwargs: Any,
461    ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]:
462        """
463        Get the library items for this mediatype.
464
465        :param favorite: Filter by favorite status.
466        :param search: Filter by search query.
467        :param limit: Maximum number of items to return.
468        :param offset: Number of items to skip.
469        :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
470        :param provider: Filter by provider instance ID (single string or list).
471        :param genre: Filter by genre id(s).
472        :param played_only: Only include items that have been played (last_played > 0).
473        :param summary: When True (default), return slim summary items containing only the
474            fields needed for a list view. Set to False to get fully hydrated items.
475        :param collapse_collections: Collapse available collections. Items in a collection won't
476            be returned individually.
477        """
478        items = await self.get_library_items_by_query(
479            favorite=favorite,
480            search=search,
481            limit=limit,
482            offset=offset,
483            order_by=order_by,
484            provider_filter=self._ensure_provider_filter(provider),
485            genre_ids=genre,
486            played_only=played_only,
487            in_library_only=True,
488            summary=summary,
489            collapse_collections=collapse_collections,
490        )
491        if (
492            kwargs.get("_localized_fallback", True)
493            and search
494            and not items
495            and self.media_type in (MediaType.GENRE, MediaType.PLAYLIST)
496        ):
497            return await self._localized_search_fallback(
498                search,
499                limit=limit,
500                offset=offset,
501                favorite=favorite,
502                order_by=order_by,
503                provider=provider,
504                genre=genre,
505                summary=summary,
506            )
507        return items
508
509    async def iter_library_items(
510        self,
511        favorite: bool | None = None,
512        search: str | None = None,
513        order_by: str = "sort_name",
514        provider: str | list[str] | None = None,
515        genre: int | list[int] | None = None,
516        library_items_only: bool = True,
517    ) -> AsyncGenerator[ItemCls]:
518        """Iterate all in-database items."""
519        limit: int = 500
520        offset: int = 0
521        if provider is not None:
522            provider_filter = provider if isinstance(provider, list) else [provider]
523        else:
524            provider_filter = None
525        while True:
526            next_items = await self.get_library_items_by_query(
527                favorite=favorite,
528                search=search,
529                genre_ids=genre,
530                limit=limit,
531                offset=offset,
532                order_by=order_by,
533                provider_filter=provider_filter,
534                in_library_only=library_items_only,
535            )
536            for item in next_items:
537                yield item
538            if len(next_items) < limit:
539                break
540            offset += limit
541
542    async def get(
543        self,
544        item_id: str,
545        provider_instance_id_or_domain: str,
546        allow_update_metadata: bool = True,
547    ) -> ItemCls:
548        """
549        Return (full) details for a single media item.
550
551        Tries to find the item in the library first, falling back to
552        fetching directly from the provider if not found.
553
554        :param item_id: The provider item id to fetch.
555        :param provider_instance_id_or_domain: The provider instance id or
556            domain to fetch the item from.
557        :param allow_update_metadata: Schedule a metadata refresh on access.
558            Set to False when fetching items in bulk (e.g. provider sync).
559        """
560        # always prefer the full library item if we have it
561        if library_item := await self.get_library_item_by_prov_id(
562            item_id,
563            provider_instance_id_or_domain,
564        ):
565            # schedule a refresh of the metadata on access of the item
566            # e.g. the item is being played or opened in the UI
567            if allow_update_metadata:
568                assert library_item.uri is not None
569                self.mass.metadata.schedule_update_metadata(library_item)
570            return library_item
571        # grab full details from the provider
572        return await self.get_provider_item(
573            item_id,
574            provider_instance_id_or_domain,
575        )
576
577    async def search(
578        self,
579        search_query: str,
580        provider_instance_id_or_domain: str,
581        limit: int = 25,
582    ) -> list[ItemCls]:
583        """Search database or provider with given query."""
584        # create safe search string
585        search_query = search_query.replace("/", " ").replace("'", "")
586        if provider_instance_id_or_domain == "library":
587            return await self.library_items(
588                search=search_query, limit=limit, summary=False, collapse_collections=False
589            )
590        if not (prov := self.mass.get_provider(provider_instance_id_or_domain)):
591            return []
592        prov = cast("MusicProvider", prov)
593        if ProviderFeature.SEARCH not in prov.supported_features:
594            return []
595        if not self.mass.music.library_supported(prov, self.media_type):
596            # assume library supported also means that this mediatype is supported
597            return []
598        searchresult = await prov.search(
599            search_query,
600            [self.media_type],
601            limit,
602        )
603        match self.media_type:
604            case MediaType.ARTIST:
605                return cast("list[ItemCls]", searchresult.artists)
606            case MediaType.ALBUM:
607                return cast("list[ItemCls]", searchresult.albums)
608            case MediaType.TRACK:
609                return cast("list[ItemCls]", searchresult.tracks)
610            case MediaType.PLAYLIST:
611                return cast("list[ItemCls]", searchresult.playlists)
612            case MediaType.AUDIOBOOK:
613                return cast("list[ItemCls]", searchresult.audiobooks)
614            case MediaType.PODCAST:
615                return cast("list[ItemCls]", searchresult.podcasts)
616            case MediaType.RADIO:
617                return cast("list[ItemCls]", searchresult.radio)
618            case _:
619                return []
620
621    async def get_collection(self, item_id: str) -> MediaCollection[ItemCls]:
622        """Get a single collection."""
623        name = get_collection_name_from_item_id(item_id)
624        query_params: dict[str, Any] = {"collection_name": name}
625        sql_query, base_query_params = self._build_final_query([], [], None, summary=False)
626        for key, value in base_query_params.items():
627            query_params.setdefault(key, value)
628        sql_query = await self._adapt_query_for_collections(
629            sql_query, query_params, summary=False, order_by=None, collection_name=name
630        )
631        db_rows = await self.mass.music.database.get_rows_from_query(
632            sql_query, query_params, limit=1, offset=0
633        )
634        if len(db_rows) != 1:
635            raise MediaNotFoundError(f"Collection {name} not found.")
636
637        return cast(
638            "MediaCollection[ItemCls]",
639            MediaCollection(
640                item_id=get_collection_item_id(db_rows[0]["name"], item_media_type=self.media_type),
641                name=db_rows[0]["name"],
642                provider="library",
643                provider_mappings=set(),
644                items=UniqueList(
645                    [
646                        self.item_cls.from_dict(self._parse_db_row(json_loads(x)))
647                        for x in json_loads(db_rows[0]["media_data"])
648                    ]
649                ),
650            ),
651        )
652
653    async def get_library_item(self, item_id: int | str) -> ItemCls:
654        """Get single library item by id."""
655        db_id = int(item_id)  # ensure integer
656        extra_query = f"WHERE {self.db_table}.item_id = :item_id"
657        for db_item in await self.get_library_items_by_query(
658            extra_query_parts=[extra_query],
659            extra_query_params={"item_id": db_id},
660            in_library_only=False,
661        ):
662            return db_item
663        msg = f"{self.media_type.value} not found in library: {db_id}"
664        raise MediaNotFoundError(msg)
665
666    async def get_library_item_by_prov_id(
667        self,
668        item_id: str,
669        provider_instance_id_or_domain: str,
670    ) -> ItemCls | None:
671        """Get the library item for the given provider item, if present."""
672        assert item_id
673        assert provider_instance_id_or_domain
674        if provider_instance_id_or_domain == "library":
675            try:
676                return await self.get_library_item(item_id)
677            except MediaNotFoundError:
678                return None
679        for item in await self.get_library_items_by_prov_id(
680            provider_instance_id_or_domain=provider_instance_id_or_domain,
681            provider_item_id=item_id,
682        ):
683            return item
684        return None
685
686    @final
687    async def get_library_item_by_prov_mappings(
688        self,
689        provider_mappings: Iterable[ProviderMapping],
690    ) -> ItemCls | None:
691        """Get the library item for the given provider_instance."""
692        # always prefer provider instance first
693        for mapping in provider_mappings:
694            for item in await self.get_library_items_by_prov_id(
695                provider_instance=mapping.provider_instance,
696                provider_item_id=mapping.item_id,
697            ):
698                return item
699        # check by domain too
700        for mapping in provider_mappings:
701            for item in await self.get_library_items_by_prov_id(
702                provider_domain=mapping.provider_domain,
703                provider_item_id=mapping.item_id,
704            ):
705                return item
706        return None
707
708    @final
709    async def get_library_item_sync_details(
710        self,
711        provider_mappings: Iterable[ProviderMapping],
712    ) -> LibraryItemSyncDetails | None:
713        """
714        Get a lightweight sync snapshot of the library item for the given provider mappings.
715
716        Returns only the scalar columns and raw provider mapping rows the library sync
717        needs for its change detection, without hydrating a full MediaItem object.
718        Resolution order matches get_library_item_by_prov_mappings (instance first,
719        then domain).
720        """
721        extra_columns, extra_joins, extra_params = self._sync_details_query_parts()
722        base_sql = f"""
723            SELECT
724                {self.db_table}.item_id,
725                {self.db_table}.favorite,
726                {self.db_table}.timestamp_added,
727                (SELECT JSON_GROUP_ARRAY(
728                    json_object(
729                        'item_id', pm.provider_item_id,
730                        'provider_domain', pm.provider_domain,
731                        'provider_instance', pm.provider_instance,
732                        'available', pm.available,
733                        'in_library', pm.in_library,
734                        'is_unique', pm.is_unique
735                    )) FROM provider_mappings pm WHERE pm.item_id = {self.db_table}.item_id
736                        AND pm.media_type = '{self.media_type.value}') AS provider_mappings
737                {extra_columns}
738            FROM {self.db_table}
739            {extra_joins}
740            WHERE {self.db_table}.item_id IN (
741                SELECT item_id FROM provider_mappings
742                WHERE provider_mappings.media_type = '{self.media_type.value}'
743                AND provider_mappings.{{prov_column}} = :prov_id
744                AND provider_mappings.provider_item_id = :prov_item_id
745            )
746        """
747        # always prefer provider instance first, then domain
748        # (same resolution order as get_library_item_by_prov_mappings)
749        for prov_column in ("provider_instance", "provider_domain"):
750            for mapping in provider_mappings:
751                for db_row in await self.mass.music.database.get_rows_from_query(
752                    base_sql.format(prov_column=prov_column),
753                    {
754                        **extra_params,
755                        "prov_id": getattr(mapping, prov_column),
756                        "prov_item_id": mapping.item_id,
757                    },
758                    limit=1,
759                ):
760                    return self._parse_sync_details_row(db_row)
761        return None
762
763    @final
764    async def get_library_item_by_external_id(
765        self, external_id: str, external_id_type: ExternalID | None = None
766    ) -> ItemCls | None:
767        """Get the library item for the given external id."""
768        subquery_parts = [
769            "media_type = :ext_id_media_type",
770            "external_id = :external_id",
771        ]
772        query_params: dict[str, Any] = {
773            "ext_id_media_type": self.media_type.value,
774            "external_id": external_id,
775        }
776        if external_id_type:
777            subquery_parts.append("external_id_type = :external_id_type")
778            query_params["external_id_type"] = str(external_id_type)
779        subquery = (
780            f"SELECT item_id FROM {DB_TABLE_EXTERNAL_ID_LOOKUP} "
781            f"WHERE {' AND '.join(subquery_parts)}"
782        )
783        query = f"{self.db_table}.item_id IN ({subquery})"
784        for item in await self.get_library_items_by_query(
785            limit=1,
786            extra_query_parts=[query],
787            extra_query_params=query_params,
788        ):
789            return item
790        return None
791
792    @final
793    async def get_library_item_by_external_ids(
794        self, external_ids: set[tuple[ExternalID, str]]
795    ) -> ItemCls | None:
796        """Get the library item for (one of) the given external ids."""
797        for external_id_type, external_id in external_ids:
798            if match := await self.get_library_item_by_external_id(external_id, external_id_type):
799                return match
800        return None
801
802    @final
803    async def get_library_items_by_prov_id(
804        self,
805        provider_domain: str | None = None,
806        provider_instance: str | None = None,
807        provider_instance_id_or_domain: str | None = None,
808        provider_item_id: str | None = None,
809        provider_item_ids: list[str] | None = None,
810        limit: int = 500,
811        offset: int = 0,
812    ) -> list[ItemCls]:
813        """
814        Fetch all records from library for given provider.
815
816        :param provider_item_ids: When given, batch-match this list of provider
817            item ids in a single query (the plural form of provider_item_id);
818            takes precedence over provider_item_id when both are passed. An
819            empty list matches nothing (distinct from None, which applies no
820            item-id filter).
821        """
822        assert provider_instance_id_or_domain != "library"
823        assert provider_domain != "library"
824        assert provider_instance != "library"
825        if provider_item_ids is not None and not provider_item_ids:
826            return []
827        subquery_parts: list[str] = []
828        query_params: dict[str, Any] = {}
829        if provider_instance:
830            query_params = {"prov_id": provider_instance}
831            subquery_parts.append("provider_mappings.provider_instance = :prov_id")
832        elif provider_domain:
833            query_params = {"prov_id": provider_domain}
834            subquery_parts.append("provider_mappings.provider_domain = :prov_id")
835        else:
836            query_params = {"prov_id": provider_instance_id_or_domain}
837            subquery_parts.append(
838                "(provider_mappings.provider_instance = :prov_id "
839                "OR provider_mappings.provider_domain = :prov_id)"
840            )
841        if provider_item_ids:
842            placeholders = ", ".join(f":item_id_{i}" for i in range(len(provider_item_ids)))
843            subquery_parts.append(f"provider_mappings.provider_item_id IN ({placeholders})")
844            for i, item_id in enumerate(provider_item_ids):
845                query_params[f"item_id_{i}"] = item_id
846        elif provider_item_id:
847            subquery_parts.append("provider_mappings.provider_item_id = :item_id")
848            query_params["item_id"] = provider_item_id
849        subquery = f"SELECT item_id FROM provider_mappings WHERE {' AND '.join(subquery_parts)}"
850        query = f"WHERE {self.db_table}.item_id IN ({subquery})"
851        return await self.get_library_items_by_query(
852            limit=limit,
853            offset=offset,
854            extra_query_parts=[query],
855            extra_query_params=query_params,
856            in_library_only=False,
857        )
858
859    @final
860    async def iter_library_items_by_prov_id(
861        self,
862        provider_instance_id_or_domain: str,
863        provider_item_id: str | None = None,
864    ) -> AsyncGenerator[ItemCls]:
865        """Iterate all records from database for given provider."""
866        limit: int = 500
867        offset: int = 0
868        while True:
869            next_items = await self.get_library_items_by_prov_id(
870                provider_instance_id_or_domain=provider_instance_id_or_domain,
871                provider_item_id=provider_item_id,
872                limit=limit,
873                offset=offset,
874            )
875            for item in next_items:
876                yield item
877            if len(next_items) < limit:
878                break
879            offset += limit
880
881    @final
882    async def set_favorite(self, item_id: str | int, favorite: bool) -> None:
883        """Set the favorite bool on a database item."""
884        db_id = int(item_id)  # ensure integer
885        library_item = await self.get_library_item(db_id)
886        if library_item.favorite == favorite:
887            return
888        match = {"item_id": db_id}
889        await self.mass.music.database.update(self.db_table, match, {"favorite": favorite})
890        library_item = await self.get_library_item(db_id)
891        self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, library_item.uri, library_item)
892
893    @guard_single_request
894    @final
895    async def get_provider_item(
896        self,
897        item_id: str,
898        provider_instance_id_or_domain: str,
899        force_refresh: bool = False,
900        fallback: ItemMapping | ItemCls | None = None,
901    ) -> ItemCls:
902        """Return item details for the given provider item id."""
903        if provider_instance_id_or_domain == "library":
904            return await self.get_library_item(item_id)
905        if not (provider := self.mass.get_provider(provider_instance_id_or_domain)):
906            raise ProviderUnavailableError(f"{provider_instance_id_or_domain} is not available")
907        if provider := self.mass.get_provider(provider_instance_id_or_domain):
908            provider = cast("MusicProvider | PluginProvider", provider)
909            with suppress(MediaNotFoundError):
910                async with self.mass.cache.handle_refresh(force_refresh):
911                    if self.media_type == MediaType.PLAYLIST:
912                        return cast("ItemCls", await provider.get_playlist(item_id))
913                    music_prov = cast("MusicProvider", provider)
914                    if self.media_type == MediaType.ARTIST:
915                        return cast("ItemCls", await music_prov.get_artist(item_id))
916                    if self.media_type == MediaType.ALBUM:
917                        return cast("ItemCls", await music_prov.get_album(item_id))
918                    if self.media_type == MediaType.TRACK:
919                        return cast("ItemCls", await music_prov.get_track(item_id))
920                    if self.media_type == MediaType.RADIO:
921                        return cast("ItemCls", await music_prov.get_radio(item_id))
922                    if self.media_type == MediaType.AUDIOBOOK:
923                        return cast("ItemCls", await music_prov.get_audiobook(item_id))
924                    if self.media_type == MediaType.PODCAST:
925                        return cast("ItemCls", await music_prov.get_podcast(item_id))
926        # if we reach this point all possibilities failed and the item could not be found.
927        # There is a possibility that the (streaming) provider changed the id of the item
928        # so we return the previous details (if we have any) marked as unavailable, so
929        # at least we have the possibility to sort out the new id through matching logic.
930        fallback = fallback or await self.get_library_item_by_prov_id(
931            item_id, provider_instance_id_or_domain
932        )
933        if (
934            fallback
935            and isinstance(fallback, ItemMapping)
936            and (fallback_provider := self.mass.get_provider(fallback.provider))
937        ):
938            # fallback is a ItemMapping, try to convert to full item
939            with suppress(LookupError, TypeError, ValueError):
940                return cast(
941                    "ItemCls",
942                    self.item_cls.from_dict(
943                        {
944                            **fallback.to_dict(),
945                            "provider_mappings": [
946                                {
947                                    "item_id": fallback.item_id,
948                                    "provider_domain": fallback_provider.domain,
949                                    "provider_instance": fallback_provider.instance_id,
950                                    "available": fallback.available,
951                                }
952                            ],
953                        }
954                    ),
955                )
956        if fallback:
957            # simply return the fallback item
958            return cast("ItemCls", fallback)
959        # all options exhausted, we really can not find this item
960        msg = (
961            f"{self.media_type.value}://{item_id} not "
962            f"found on provider {provider_instance_id_or_domain}"
963        )
964        raise MediaNotFoundError(msg)
965
966    @final
967    async def add_provider_mapping(
968        self, item_id: str | int, provider_mapping: ProviderMapping
969    ) -> None:
970        """Add provider mapping to existing library item."""
971        await self.add_provider_mappings(item_id, [provider_mapping])
972
973    @final
974    async def add_provider_mappings(
975        self, item_id: str | int, provider_mappings: Iterable[ProviderMapping]
976    ) -> None:
977        """
978        Add provider mappings to existing library item.
979
980        :param item_id: The library item ID to add mappings to.
981        :param provider_mappings: The provider mappings to add.
982        """
983        db_id = int(item_id)  # ensure integer
984        library_item = await self.get_library_item(db_id)
985        new_mappings: set[ProviderMapping] = set()
986        for provider_mapping in provider_mappings:
987            # ignore if the mapping is already present
988            if provider_mapping not in library_item.provider_mappings:
989                new_mappings.add(provider_mapping)
990        if not new_mappings:
991            return
992        # handle special case where the user wants to merge 2 library items
993        for mapping in new_mappings:
994            if _library_item := await self.get_library_item_by_prov_id(
995                mapping.item_id, mapping.provider_instance
996            ):
997                if _library_item.item_id != library_item.item_id:
998                    # merging items
999                    self.logger.debug(
1000                        "merging item id %s into item id %s based on provider mapping %s/%s",
1001                        _library_item.item_id,
1002                        library_item.item_id,
1003                        mapping.provider_instance,
1004                        mapping.item_id,
1005                    )
1006                    await self.remove_item_from_library(_library_item.item_id, recursive=True)
1007                    break
1008        library_item.provider_mappings.update(new_mappings)
1009        self.mass.music.match_provider_instances(library_item)
1010        await self.set_provider_mappings(db_id, library_item.provider_mappings)
1011        self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, library_item.uri, library_item)
1012
1013    @final
1014    async def update_provider_mapping(
1015        self,
1016        item_id: str | int,
1017        provider_instance_id: str,
1018        provider_item_id: str,
1019        *,
1020        available: bool | Any = UNSET,
1021        in_library: bool | Any = UNSET,
1022        is_unique: bool | None | Any = UNSET,
1023        url: str | None | Any = UNSET,
1024        details: str | None | Any = UNSET,
1025        audio_format: AudioFormat | Any = UNSET,
1026    ) -> None:
1027        """Update an existing provider mapping for a library item."""
1028        db_id = int(item_id)  # ensure integer
1029        library_item = await self.get_library_item(db_id)
1030
1031        # find the current mapping (strictly by provider instance + provider item id)
1032        cur_mapping: ProviderMapping | None = None
1033        for mapping in library_item.provider_mappings:
1034            if (
1035                mapping.provider_instance == provider_instance_id
1036                and mapping.item_id == provider_item_id
1037            ):
1038                cur_mapping = mapping
1039                break
1040        if cur_mapping is None:
1041            msg = (
1042                f"Provider mapping {provider_instance_id}/{provider_item_id} "
1043                f"not found for item {db_id}"
1044            )
1045            raise MediaNotFoundError(msg)
1046
1047        # guard against nulls for NOT NULL columns
1048        if available is None:
1049            available = UNSET
1050        if in_library is None:
1051            in_library = UNSET
1052
1053        updates: dict[str, Any] = {}
1054        if available is not UNSET:
1055            updates["available"] = bool(available)
1056        if in_library is not UNSET:
1057            updates["in_library"] = bool(in_library)
1058        if is_unique is not UNSET:
1059            updates["is_unique"] = is_unique
1060        if url is not UNSET:
1061            updates["url"] = url
1062        if details is not UNSET:
1063            updates["details"] = details
1064        if audio_format is not UNSET:
1065            updates["audio_format"] = serialize_to_json(audio_format)
1066
1067        if not updates:
1068            return
1069
1070        match = {
1071            "media_type": self.media_type.value,
1072            "item_id": db_id,
1073            "provider_instance": provider_instance_id,
1074            "provider_item_id": provider_item_id,
1075        }
1076        await self.mass.music.database.update(DB_TABLE_PROVIDER_MAPPINGS, match, updates)
1077
1078        # Re-fetch the updated item so the event payload reflects persisted DB state.
1079        updated_item = await self.get_library_item(db_id)
1080        self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, updated_item.uri, updated_item)
1081
1082    @final
1083    async def remove_provider_mapping(
1084        self, item_id: str | int, provider_instance_id: str, provider_item_id: str
1085    ) -> None:
1086        """Remove provider mapping(s) from item."""
1087        db_id = int(item_id)  # ensure integer
1088        try:
1089            library_item = await self.get_library_item(db_id)
1090        except MediaNotFoundError:
1091            # edge case: already deleted / race condition
1092            return
1093
1094        remaining_mappings = {
1095            x
1096            for x in library_item.provider_mappings
1097            if not (x.provider_instance == provider_instance_id and x.item_id == provider_item_id)
1098        }
1099        if not remaining_mappings:
1100            # this was the last mapping, so remove the entire library item, which also
1101            # clears its provider mapping rows. Dropping those rows up front would leave
1102            # the item behind without any mappings if the removal itself fails.
1103            with suppress(MediaNotFoundError):
1104                await self.remove_item_from_library(db_id)
1105            return
1106
1107        # update provider_mappings table
1108        await self.mass.music.database.delete(
1109            DB_TABLE_PROVIDER_MAPPINGS,
1110            {
1111                "media_type": self.media_type.value,
1112                "item_id": db_id,
1113                "provider_instance": provider_instance_id,
1114                "provider_item_id": provider_item_id,
1115            },
1116        )
1117        # cleanup playlog table
1118        await self.mass.music.database.delete(
1119            DB_TABLE_PLAYLOG,
1120            {
1121                "media_type": self.media_type.value,
1122                "item_id": provider_item_id,
1123                "provider": provider_instance_id,
1124            },
1125        )
1126        library_item.provider_mappings = remaining_mappings
1127        # if this was the last mapping for the provider instance, strip any artwork
1128        # that belonged to it (e.g. local file paths that are no longer resolvable)
1129        images_changed = not any(
1130            x.provider_instance == provider_instance_id for x in remaining_mappings
1131        ) and await self._remove_provider_images(db_id, provider_instance_id)
1132        self.logger.debug(
1133            "removed provider_mapping %s/%s from item id %s",
1134            provider_instance_id,
1135            provider_item_id,
1136            db_id,
1137        )
1138        # the removed provider mapping is itself a change to the item, so always notify
1139        # (unless suppressed during a bulk cleanup); re-fetch first when images were
1140        # stripped so the event payload stays accurate
1141        if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
1142            event_item = await self.get_library_item(db_id) if images_changed else library_item
1143            self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, event_item.uri, event_item)
1144
1145    @final
1146    async def remove_provider_mappings(self, item_id: str | int, provider_instance_id: str) -> None:
1147        """Remove all provider mappings from an item."""
1148        db_id = int(item_id)  # ensure integer
1149        try:
1150            library_item = await self.get_library_item(db_id)
1151        except MediaNotFoundError:
1152            # edge case: already deleted / race condition, just drop any leftover rows
1153            await self.mass.music.database.delete(
1154                DB_TABLE_PROVIDER_MAPPINGS,
1155                {
1156                    "media_type": self.media_type.value,
1157                    "item_id": db_id,
1158                    "provider_instance": provider_instance_id,
1159                },
1160            )
1161            return
1162
1163        remaining_mappings = {
1164            x for x in library_item.provider_mappings if x.provider_instance != provider_instance_id
1165        }
1166        if not remaining_mappings:
1167            # these were the last mappings, so remove the entire library item, which also
1168            # clears its provider mapping rows. Dropping those rows up front would leave
1169            # the item behind without any mappings if the removal itself fails.
1170            with suppress(MediaNotFoundError):
1171                await self.remove_item_from_library(db_id)
1172            return
1173
1174        # update provider_mappings table
1175        await self.mass.music.database.delete(
1176            DB_TABLE_PROVIDER_MAPPINGS,
1177            {
1178                "media_type": self.media_type.value,
1179                "item_id": db_id,
1180                "provider_instance": provider_instance_id,
1181            },
1182        )
1183        library_item.provider_mappings = remaining_mappings
1184        # the item is kept (it still has other providers), but it may carry artwork
1185        # that belonged to the removed provider (e.g. local file paths that are no
1186        # longer resolvable), so strip those images from the stored metadata
1187        images_changed = await self._remove_provider_images(db_id, provider_instance_id)
1188        self.logger.debug(
1189            "removed all provider mappings for provider %s from item id %s",
1190            provider_instance_id,
1191            db_id,
1192        )
1193        # the removed provider mapping(s) are themselves a change to the item, so
1194        # always notify (unless suppressed during a bulk cleanup); re-fetch first when
1195        # images were stripped so the event payload stays accurate
1196        if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
1197            event_item = await self.get_library_item(db_id) if images_changed else library_item
1198            self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, event_item.uri, event_item)
1199
1200    @final
1201    async def set_provider_mappings(
1202        self,
1203        item_id: str | int,
1204        provider_mappings: Iterable[ProviderMapping],
1205        overwrite: bool = False,
1206    ) -> None:
1207        """
1208        Update the provider_mappings table for the media item.
1209
1210        An empty set of mappings never clears the stored rows: an item without any
1211        mapping can not be played or resolved.
1212        """
1213        db_id = int(item_id)  # ensure integer
1214        prov_map_objs: list[dict[str, Any]] = []
1215        for provider_mapping in provider_mappings:
1216            prov_map_obj = {
1217                "media_type": self.media_type.value,
1218                "item_id": db_id,
1219                "provider_domain": provider_mapping.provider_domain,
1220                "provider_instance": provider_mapping.provider_instance,
1221                "provider_item_id": provider_mapping.item_id,
1222                "available": provider_mapping.available,
1223                "audio_format": serialize_to_json(provider_mapping.audio_format),
1224            }
1225            for key in ("url", "details", "in_library", "is_unique"):
1226                if (value := getattr(provider_mapping, key, None)) is not None:
1227                    prov_map_obj[key] = value
1228            prov_map_objs.append(prov_map_obj)
1229        if not prov_map_objs:
1230            if overwrite:
1231                # a caller asking to replace all mappings with none is a bug,
1232                # so keep the stored rows and make the attempt visible
1233                self.logger.warning(
1234                    "Ignoring request to clear all provider mappings of %s item id %s",
1235                    self.media_type.value,
1236                    db_id,
1237                )
1238            return
1239        if overwrite:
1240            # on overwrite, clear the provider_mappings table first
1241            # this is done for filesystem provider changing the path (and thus item_id)
1242            await self.mass.music.database.delete(
1243                DB_TABLE_PROVIDER_MAPPINGS,
1244                {"media_type": self.media_type.value, "item_id": db_id},
1245            )
1246        await self.mass.music.database.upsert_many(
1247            DB_TABLE_PROVIDER_MAPPINGS,
1248            prov_map_objs,
1249        )
1250
1251    @final
1252    async def set_external_ids(
1253        self,
1254        item_id: str | int,
1255        external_ids: Iterable[tuple[ExternalID, str]],
1256    ) -> None:
1257        """Update the external_id_lookup table rows for the media item."""
1258        db_id = int(item_id)  # ensure integer
1259        await self.mass.music.database.delete(
1260            DB_TABLE_EXTERNAL_ID_LOOKUP,
1261            {"media_type": self.media_type.value, "item_id": db_id},
1262        )
1263        if lookup_rows := [
1264            {
1265                "media_type": self.media_type.value,
1266                "external_id_type": external_id_type,
1267                "external_id": external_id,
1268                "item_id": db_id,
1269            }
1270            for external_id_type, external_id in external_ids
1271        ]:
1272            await self.mass.music.database.upsert_many(DB_TABLE_EXTERNAL_ID_LOOKUP, lookup_rows)
1273
1274    @abstractmethod
1275    async def match_providers(self, db_item: ItemCls) -> None:
1276        """
1277        Try to find match on all (streaming) providers for the provided (database) item.
1278
1279        This is used to link objects of different providers/qualities together.
1280        """
1281
1282    if TYPE_CHECKING:
1283
1284        @overload
1285        async def get_library_items_by_query(
1286            self,
1287            favorite: bool | None = None,
1288            search: str | None = None,
1289            limit: int = 500,
1290            offset: int = 0,
1291            order_by: str | None = None,
1292            provider_filter: list[str] | None = None,
1293            extra_query_parts: list[str] | None = None,
1294            extra_query_params: dict[str, Any] | None = None,
1295            extra_join_parts: list[str] | None = None,
1296            genre_ids: int | list[int] | None = None,
1297            played_only: bool = False,
1298            in_library_only: bool = False,
1299            summary: bool = False,
1300            *,
1301            collapse_collections: Literal[True],
1302        ) -> list[ItemCls | MediaCollection[ItemCls]]: ...
1303
1304        @overload
1305        async def get_library_items_by_query(
1306            self,
1307            favorite: bool | None = None,
1308            search: str | None = None,
1309            limit: int = 500,
1310            offset: int = 0,
1311            order_by: str | None = None,
1312            provider_filter: list[str] | None = None,
1313            extra_query_parts: list[str] | None = None,
1314            extra_query_params: dict[str, Any] | None = None,
1315            extra_join_parts: list[str] | None = None,
1316            genre_ids: int | list[int] | None = None,
1317            played_only: bool = False,
1318            in_library_only: bool = False,
1319            summary: bool = False,
1320            *,
1321            collapse_collections: Literal[False] = False,
1322        ) -> list[ItemCls]: ...
1323
1324        @overload
1325        async def get_library_items_by_query(
1326            self,
1327            favorite: bool | None = None,
1328            search: str | None = None,
1329            limit: int = 500,
1330            offset: int = 0,
1331            order_by: str | None = None,
1332            provider_filter: list[str] | None = None,
1333            extra_query_parts: list[str] | None = None,
1334            extra_query_params: dict[str, Any] | None = None,
1335            extra_join_parts: list[str] | None = None,
1336            genre_ids: int | list[int] | None = None,
1337            played_only: bool = False,
1338            in_library_only: bool = False,
1339            summary: bool = False,
1340            *,
1341            collapse_collections: bool,
1342        ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]: ...
1343
1344    @final
1345    async def get_library_items_by_query(  # noqa: PLR0913
1346        self,
1347        favorite: bool | None = None,
1348        search: str | None = None,
1349        limit: int = 500,
1350        offset: int = 0,
1351        order_by: str | None = None,
1352        provider_filter: list[str] | None = None,
1353        extra_query_parts: list[str] | None = None,
1354        extra_query_params: dict[str, Any] | None = None,
1355        extra_join_parts: list[str] | None = None,
1356        genre_ids: int | list[int] | None = None,
1357        played_only: bool = False,
1358        in_library_only: bool = False,
1359        summary: bool = False,
1360        *,
1361        collapse_collections: bool = False,
1362    ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]:
1363        """Fetch MediaItem records from database by building the query."""
1364        query_params = dict(extra_query_params) if extra_query_params else {}
1365        query_parts: list[str] = list(extra_query_parts) if extra_query_parts else []
1366        join_parts: list[str] = list(extra_join_parts) if extra_join_parts else []
1367        search = self._preprocess_search(search)
1368        genre_ids = self._preprocess_genre_ids(genre_ids)
1369        # create special performant random query
1370        if order_by and order_by.startswith("random"):
1371            self._apply_random_subquery(
1372                query_parts=query_parts,
1373                query_params=query_params,
1374                join_parts=join_parts,
1375                favorite=favorite,
1376                search=search if not collapse_collections else None,
1377                genre_ids=genre_ids,
1378                provider_filter=provider_filter,
1379                played_only=played_only,
1380                limit=limit,
1381                in_library_only=in_library_only,
1382            )
1383        else:
1384            # apply filters
1385            self._apply_filters(
1386                query_parts=query_parts,
1387                query_params=query_params,
1388                favorite=favorite,
1389                search=search if not collapse_collections else None,
1390                genre_ids=genre_ids,
1391                provider_filter=provider_filter,
1392                played_only=played_only,
1393                in_library_only=in_library_only,
1394            )
1395        # build and execute final query
1396        sql_query, base_query_params = self._build_final_query(
1397            query_parts, join_parts, order_by, summary=summary
1398        )
1399        # base query params act as defaults: callers may override them via extra_query_params
1400        for key, value in base_query_params.items():
1401            query_params.setdefault(key, value)
1402
1403        if collapse_collections:
1404            if search:
1405                query_params["search"] = f"%{search}%"
1406            sql_query = await self._adapt_query_for_collections(
1407                sql_query, query_params, summary=summary, order_by=order_by, search=search
1408            )
1409
1410        db_rows = await self.mass.music.database.get_rows_from_query(
1411            sql_query, query_params, limit=limit, offset=offset
1412        )
1413        if collapse_collections:
1414            items: list[ItemCls | MediaCollection[ItemCls]] = []
1415
1416            def _parse_method(x: str) -> ItemCls:
1417                if summary:
1418                    return cast("ItemCls", self._parse_summary_row(json_loads(x)))
1419                return cast(
1420                    "ItemCls",
1421                    self.item_cls.from_dict(self._parse_db_row(json_loads(x))),
1422                )
1423
1424            for db_row in db_rows:
1425                if db_row["type"] == "single":
1426                    items.append(_parse_method(db_row["media_data"]))
1427                elif db_row["type"] == "collection":
1428                    items.append(
1429                        MediaCollection[ItemCls](
1430                            item_id=get_collection_item_id(
1431                                db_row["name"], item_media_type=self.media_type
1432                            ),
1433                            name=db_row["name"],
1434                            provider="library",
1435                            provider_mappings=set(),
1436                            items=UniqueList(
1437                                [_parse_method(x) for x in json_loads(db_row["media_data"])]
1438                            ),
1439                        )
1440                    )
1441            return items
1442        if summary:
1443            return [cast("ItemCls", self._parse_summary_row(db_row)) for db_row in db_rows]
1444        return [
1445            cast("ItemCls", self.item_cls.from_dict(self._parse_db_row(db_row)))
1446            for db_row in db_rows
1447        ]
1448
1449    @final
1450    async def _get_library_item_by_match(self, item: ItemCls | ItemMapping) -> int | None:
1451        if item.provider == "library":
1452            return int(item.item_id)
1453        # search by provider mappings if item is ItemMapping
1454        if isinstance(item, ItemMapping):
1455            if cur_item := await self.get_library_item_by_prov_id(item.item_id, item.provider):
1456                return int(cur_item.item_id)
1457
1458        # for all other items that are MediaItemType, check provider_mappings if it exists
1459        provider_mappings = getattr(item, "provider_mappings", None)
1460        if provider_mappings:
1461            if cur_item := await self.get_library_item_by_prov_mappings(provider_mappings):
1462                return int(cur_item.item_id)
1463        if cur_item := await self.get_library_item_by_external_ids(item.external_ids):
1464            # existing item match by external id
1465            # Double check external IDs - if MBID exists, regards that as overriding
1466            if compare_media_item(item, cur_item):
1467                return int(cur_item.item_id)
1468        # search by (exact) name match
1469        query = f"{self.db_table}.name = :name OR {self.db_table}.sort_name = :sort_name"
1470        query_params = {"name": item.name, "sort_name": item.sort_name}
1471        for db_item in await self.get_library_items_by_query(
1472            extra_query_parts=[query], extra_query_params=query_params
1473        ):
1474            if compare_media_item(db_item, item, True):
1475                return int(db_item.item_id)
1476        return None
1477
1478    def _external_ids_query(
1479        self, media_type: MediaType | None = None, table_alias: str | None = None
1480    ) -> str:
1481        """
1482        Return a subquery that selects the external ids of a media item as a JSON array.
1483
1484        :param media_type: Media type to select the external ids for, defaults to
1485            this controller's media type.
1486        :param table_alias: (Aliased) table name the subquery correlates against,
1487            defaults to this controller's table.
1488        """
1489        media_type = media_type or self.media_type
1490        table_alias = table_alias or self.db_table
1491        return (
1492            f"(SELECT JSON_GROUP_ARRAY(json_array("
1493            f"{DB_TABLE_EXTERNAL_ID_LOOKUP}.external_id_type, "
1494            f"{DB_TABLE_EXTERNAL_ID_LOOKUP}.external_id)) "
1495            f"FROM {DB_TABLE_EXTERNAL_ID_LOOKUP} "
1496            f"WHERE {DB_TABLE_EXTERNAL_ID_LOOKUP}.media_type = '{media_type.value}' "
1497            f"AND {DB_TABLE_EXTERNAL_ID_LOOKUP}.item_id = {table_alias}.item_id)"
1498        )
1499
1500    def _provider_mappings_query(self) -> str:
1501        """Return a subquery that selects the provider mappings of a media item as a JSON array."""
1502        return f"""(SELECT JSON_GROUP_ARRAY(
1503            json_object(
1504                'item_id', pm.provider_item_id,
1505                'provider_domain', pm.provider_domain,
1506                'provider_instance', pm.provider_instance,
1507                'available', pm.available,
1508                'audio_format', json(pm.audio_format),
1509                'url', pm.url,
1510                'details', pm.details,
1511                'in_library', pm.in_library,
1512                'is_unique', pm.is_unique
1513            )) FROM {DB_TABLE_PROVIDER_MAPPINGS} pm
1514            WHERE pm.item_id = {self.db_table}.item_id
1515            AND pm.media_type = '{self.media_type.value}')"""
1516
1517    def _artist_mappings_summary_query(
1518        self, m2m_table: str, m2m_key: str, include_artist_type: bool = False
1519    ) -> str:
1520        """
1521        Return a subquery selecting the slim artist mappings JSON of a summary row.
1522
1523        :param m2m_table: The many-to-many table linking artists to this media type.
1524        :param m2m_key: The column in the m2m table referencing this media type's item id.
1525        :param include_artist_type: Also select the artist_type of each artist.
1526        """
1527        artist_type_part = ",\n                'artist_type', artists.artist_type"
1528        return f"""(SELECT JSON_GROUP_ARRAY(
1529            json_object(
1530                'item_id', artists.item_id,
1531                'name', artists.name,
1532                'sort_name', artists.sort_name{artist_type_part if include_artist_type else ""}
1533            )) FROM artists
1534            JOIN {m2m_table} ON artists.item_id = {m2m_table}.artist_id
1535            WHERE {m2m_table}.{m2m_key} = {self.db_table}.item_id)"""
1536
1537    def _summary_base_columns(self) -> str:
1538        """Return the SELECT columns shared by every summary query."""
1539        # the search/sort/statistics columns are selected so ORDER BY (see SORT_KEYS)
1540        # resolves them from the result set, like the full query's SELECT * does
1541        return f"""
1542            {self.db_table}.item_id,
1543            {self.db_table}.name,
1544            {self.db_table}.sort_name,
1545            {self.db_table}.favorite,
1546            {self.db_table}.search_name AS search_name,
1547            {self.db_table}.search_sort_name AS search_sort_name,
1548            {self.db_table}.play_count AS play_count,
1549            {self.db_table}.last_played AS last_played,
1550            {self.db_table}.timestamp_added AS timestamp_added,
1551            {self.db_table}.timestamp_modified AS timestamp_modified,
1552            json_extract({self.db_table}.metadata, '$.images') AS images,
1553            json_extract({self.db_table}.metadata, '$.collections') AS collections"""
1554
1555    async def _localized_search_fallback(
1556        self, search_query: str, limit: int, offset: int = 0, **call_kwargs: Any
1557    ) -> list[ItemCls]:
1558        """
1559        Retry a library search using the canonical names behind a localized query.
1560
1561        For genre/playlist searches that return nothing literally, reverse-resolve the query to the
1562        canonical (English) names of matching localized items and search those, so an item is
1563        findable by the localized name the user sees. The caller's other filters (favorite,
1564        order_by, provider and any controller-specific kwargs) are forwarded unchanged so the retry
1565        behaves like the literal search; results are merged, de-duplicated and paginated here. See
1566        ``TranslationController.reverse_lookup_media_names``.
1567        """
1568        seen: set[Any] = set()
1569        merged: list[ItemCls] = []
1570        # iterate the canonical names in a stable order, and fetch each from the start so the
1571        # offset/limit window can be applied to the merged, de-duplicated result set
1572        for name in sorted(await self.mass.translations.reverse_lookup_media_names(search_query)):
1573            for item in await self.library_items(
1574                search=name,
1575                limit=limit + offset,
1576                offset=0,
1577                _localized_fallback=False,
1578                **call_kwargs,
1579            ):
1580                if item.item_id not in seen:
1581                    seen.add(item.item_id)
1582                    merged.append(item)
1583        return merged[offset : offset + limit]
1584
1585    @abstractmethod
1586    async def _add_library_item(
1587        self,
1588        item: ItemCls,
1589        overwrite_existing: bool = False,
1590    ) -> int:
1591        """Add item to library and return the database id."""
1592
1593    @abstractmethod
1594    async def _update_library_item(
1595        self, item_id: str | int, update: ItemCls, overwrite: bool = False
1596    ) -> None:
1597        """Update existing library record in the database."""
1598
1599    def _search_filter_clause(self, search: str, query_params: dict[str, Any]) -> str:
1600        """Return the SQL WHERE clause fragment used for search filtering."""
1601        return search_name_match_clause(self.db_table, search, "search", query_params)
1602
1603    @final
1604    def _preprocess_search(self, search: str | None) -> str | None:
1605        """Normalize the search string for use in the search filter clauses."""
1606        return create_safe_string(search, True, True) if search else search
1607
1608    @final
1609    @staticmethod
1610    def _preprocess_genre_ids(genre_ids: int | list[int] | None) -> list[int] | None:
1611        if genre_ids is None:
1612            return None
1613        if isinstance(genre_ids, list):
1614            normalized = [int(x) for x in genre_ids]
1615        else:
1616            normalized = [int(genre_ids)]
1617        return normalized or None
1618
1619    @final
1620    @staticmethod
1621    def _clean_query_parts(query_parts: list[str]) -> list[str]:
1622        """Clean the query parts list by removing duplicate where statements."""
1623        return [x[5:] if x.lower().startswith("where ") else x for x in query_parts]
1624
1625    @final
1626    def _apply_random_subquery(
1627        self,
1628        query_parts: list[str],
1629        query_params: dict[str, Any],
1630        join_parts: list[str],
1631        favorite: bool | None,
1632        search: str | None,
1633        genre_ids: list[int] | None,
1634        provider_filter: list[str] | None,
1635        played_only: bool = False,
1636        limit: int = 500,
1637        in_library_only: bool = False,
1638    ) -> None:
1639        """Build a fast random subquery with all filters applied."""
1640        sub_query_parts = query_parts.copy()
1641        sub_join_parts = join_parts.copy()
1642
1643        # Apply all filters to the subquery
1644        self._apply_filters(
1645            query_parts=sub_query_parts,
1646            query_params=query_params,
1647            favorite=favorite,
1648            search=search,
1649            genre_ids=genre_ids,
1650            provider_filter=provider_filter,
1651            played_only=played_only,
1652            in_library_only=in_library_only,
1653        )
1654
1655        # Build the subquery
1656        sub_query = f"SELECT {self.db_table}.item_id FROM {self.db_table}"
1657
1658        if sub_join_parts:
1659            sub_query += f" {' '.join(sub_join_parts)}"
1660
1661        if sub_query_parts:
1662            sub_query += " WHERE " + " AND ".join(self._clean_query_parts(sub_query_parts))
1663
1664        sub_query += f" ORDER BY RANDOM() LIMIT {limit}"
1665
1666        # The query now only consists of the random subquery, which applies all filters
1667        # within itself
1668        query_parts.clear()
1669        query_parts.append(f"{self.db_table}.item_id in ({sub_query})")
1670        join_parts.clear()
1671
1672    @final
1673    def _apply_filters(
1674        self,
1675        query_parts: list[str],
1676        query_params: dict[str, Any],
1677        favorite: bool | None,
1678        search: str | None,
1679        genre_ids: list[int] | None,
1680        provider_filter: list[str] | None,
1681        played_only: bool = False,
1682        in_library_only: bool = False,
1683    ) -> None:
1684        """Apply search, favorite, and provider filters."""
1685        # handle search
1686        if search:
1687            query_parts.append(self._search_filter_clause(search, query_params))
1688        # handle favorite filter
1689        if favorite is not None:
1690            query_parts.append(f"{self.db_table}.favorite = :favorite")
1691            query_params["favorite"] = favorite
1692        # handle played_only filter
1693        if played_only:
1694            query_parts.append(f"{self.db_table}.last_played > 0")
1695        # handle genre filter
1696        if genre_ids:
1697            query_params["genre_ids"] = genre_ids
1698            query_params["genre_media_type"] = self.media_type.value
1699            query_parts.append(
1700                f"EXISTS("
1701                f"SELECT 1 FROM {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING} gm "
1702                f"WHERE gm.media_id = {self.db_table}.item_id "
1703                "AND gm.media_type = :genre_media_type "
1704                "AND gm.genre_id IN :genre_ids)"
1705            )
1706        # Apply the provider filter
1707        if provider_filter or in_library_only:
1708            query_parts.append(
1709                self._provider_filter_clause(query_params, provider_filter, in_library_only)
1710            )
1711
1712    @final
1713    def _provider_filter_clause(
1714        self,
1715        query_params: dict[str, Any],
1716        provider_filter: list[str] | None,
1717        in_library_only: bool = False,
1718    ) -> str:
1719        """
1720        Return the SQL clause that restricts items by their provider mappings.
1721
1722        At least one of provider_filter/in_library_only must be set, otherwise the
1723        returned clause only asserts that the item has any mapping at all.
1724
1725        :param query_params: Query params dict; the clause's bound params are added to it.
1726        :param provider_filter: Only match items mapped to one of these provider instances.
1727        :param in_library_only: Only match provider mappings that are in the provider's library.
1728        """
1729        # NOTE: provider mapping filters are applied as a correlated EXISTS subquery
1730        # instead of a JOIN + GROUP BY, so SQLite can stream results straight from the
1731        # sort index instead of materializing/sorting the whole (deduped) result set.
1732        query_params["provider_media_type"] = self.media_type.value
1733        conditions = [
1734            f"provider_mappings.item_id = {self.db_table}.item_id",
1735            "provider_mappings.media_type = :provider_media_type",
1736        ]
1737        if in_library_only:
1738            conditions.append("provider_mappings.in_library = 1")
1739        if provider_filter:
1740            provider_conditions = []
1741            for idx, prov in enumerate(provider_filter):
1742                param_name = f"provider_filter_{idx}"
1743                provider_conditions.append(f"provider_mappings.provider_instance = :{param_name}")
1744                query_params[param_name] = prov
1745            conditions.append(f"({' OR '.join(provider_conditions)})")
1746        return f"EXISTS(SELECT 1 FROM provider_mappings WHERE {' AND '.join(conditions)})"
1747
1748    @final
1749    def _build_final_query(
1750        self,
1751        query_parts: list[str],
1752        join_parts: list[str],
1753        order_by: str | None,
1754        summary: bool = False,
1755    ) -> tuple[str, dict[str, Any]]:
1756        """Build the final SQL query string and its (base) bound query params."""
1757        sql_query, base_query_params = self.summary_query if summary else self.base_query
1758
1759        # Add joins
1760        if join_parts:
1761            sql_query += f" {' '.join(join_parts)} "
1762
1763        # Add where clauses
1764        if query_parts:
1765            # prevent duplicate where statement
1766            sql_query += " WHERE " + " AND ".join(self._clean_query_parts(query_parts))
1767
1768        # Add grouping (only needed when caller-provided joins can fan out rows)
1769        # and ordering. Without a GROUP BY, SQLite can stream results directly
1770        # from the sort index instead of sorting the whole result set.
1771        if join_parts:
1772            sql_query += f" GROUP BY {self.db_table}.item_id"
1773
1774        if order_by:
1775            if sort_key := SORT_KEYS.get(order_by):
1776                sql_query += f" ORDER BY {sort_key}"
1777
1778        return sql_query, base_query_params
1779
1780    @final
1781    @staticmethod
1782    def _parse_db_row(db_row: Mapping[str, Any]) -> dict[str, Any]:
1783        """Parse raw db Mapping into a dict."""
1784        db_row_dict = dict(db_row)
1785        db_row_dict["provider"] = "library"
1786        db_row_dict["favorite"] = bool(db_row_dict["favorite"])
1787        db_row_dict["item_id"] = str(db_row_dict["item_id"])
1788        db_row_dict["date_added"] = datetime.fromtimestamp(
1789            db_row_dict["timestamp_added"], tz=UTC
1790        ).isoformat()
1791
1792        for key in JSON_KEYS:
1793            if key not in db_row_dict:
1794                continue
1795            if not (raw_value := db_row_dict[key]):
1796                continue
1797            db_row_dict[key] = json_loads(raw_value)
1798
1799        # parse "fully_played" as bool if present in the row
1800        if "fully_played" in db_row_dict:
1801            db_row_dict["fully_played"] = parse_optional_bool(db_row_dict["fully_played"])
1802
1803        # copy track_album --> album
1804        if track_album := db_row_dict.get("track_album"):
1805            db_row_dict["album"] = track_album
1806            db_row_dict["disc_number"] = track_album["disc_number"]
1807            db_row_dict["track_number"] = track_album["track_number"]
1808            # always prefer album image over track image
1809            if (album_images := track_album.get("images")) and (
1810                album_thumb := next((x for x in album_images if x["type"] == "thumb"), None)
1811            ):
1812                # copy album image to itemmapping single image (on the track)
1813                db_row_dict["image"] = album_thumb
1814                # also set image on the album dict for ItemMapping compatibility
1815                track_album["image"] = album_thumb
1816                if db_row_dict["metadata"].get("images"):
1817                    # merge album image with existing images
1818                    db_row_dict["metadata"]["images"] = [
1819                        album_thumb,
1820                        *db_row_dict["metadata"]["images"],
1821                    ]
1822                else:
1823                    db_row_dict["metadata"]["images"] = [album_thumb]
1824
1825        if audiobook_artists := db_row_dict.get("audiobook_artists"):
1826            _narrators = []
1827            _authors = []
1828            for artist in audiobook_artists:
1829                artist_type = artist.get("artist_type")
1830                if artist_type == "author":
1831                    _authors.append(artist)
1832                elif artist_type == "narrator":
1833                    _narrators.append(artist)
1834            if _authors:
1835                # prevent overwriting string values
1836                db_row_dict["authors"] = _authors
1837            if _narrators:
1838                # prevent overwriting string values
1839                db_row_dict["narrators"] = _narrators
1840
1841        return db_row_dict
1842
1843    @final
1844    def _ensure_provider_filter(
1845        self,
1846        provider: str | list[str] | None,
1847    ) -> list[str] | None:
1848        """Ensure the provider filter respects the current user's provider filter."""
1849        # Apply user provider filter if needed
1850        user = get_current_user()
1851        user_provider_filter = user.provider_filter if user and user.provider_filter else None
1852        final_provider_filter: list[str] | None = None
1853        if user_provider_filter:
1854            plugin_provider_instances = {
1855                prov.instance_id for prov in self.mass.providers if prov.type == ProviderType.PLUGIN
1856            }
1857            # User has a provider filter set
1858            if provider:
1859                # Explicit provider filter provided - validate against user's allowed providers
1860                requested_providers = [provider] if isinstance(provider, str) else provider
1861                # Only restrict access to music providers.
1862                final_provider_filter = [
1863                    p
1864                    for p in requested_providers
1865                    if p in user_provider_filter or p in plugin_provider_instances
1866                ]
1867                if not final_provider_filter:
1868                    # No overlap - user requested providers they don't have access to
1869                    raise InsufficientPermissions(
1870                        "User does not have permission to access the requested provider(s)."
1871                    )
1872            else:
1873                # No explicit filter - apply user music provider filter but keep plugin providers.
1874                final_provider_filter = list(
1875                    dict.fromkeys([*user_provider_filter, *plugin_provider_instances])
1876                )
1877        elif provider is not None:
1878            # No user filter - use the provided filter as is
1879            final_provider_filter = [provider] if isinstance(provider, str) else provider
1880        return final_provider_filter
1881
1882    @final
1883    def _select_provider_id(self, library_item: ItemCls) -> tuple[str, str]:
1884        """Select the correct provider id to use for fetching the item."""
1885        if not library_item.provider_mappings:
1886            msg = (
1887                f"{self.media_type.value} {library_item.item_id} "
1888                "is no longer available on any provider"
1889            )
1890            raise MediaNotFoundError(msg)
1891        user = get_current_user()
1892        user_provider_filter = user.provider_filter if user and user.provider_filter else None
1893        if not user_provider_filter:
1894            mapping = next(iter(library_item.provider_mappings))
1895            return (mapping.provider_instance, mapping.item_id)
1896
1897        # First prefer music provider mappings that are explicitly allowed for this user.
1898        # prefer user provider filter if available
1899        for mapping in library_item.provider_mappings:
1900            provider = self.mass.get_provider(mapping.provider_instance)
1901            if provider and provider.type == ProviderType.MUSIC:
1902                if mapping.provider_instance in user_provider_filter:
1903                    return (mapping.provider_instance, mapping.item_id)
1904
1905        # If no allowed music mapping exists, fall back to plugin mappings.
1906        for mapping in library_item.provider_mappings:
1907            provider = self.mass.get_provider(mapping.provider_instance)
1908            if provider and provider.type == ProviderType.PLUGIN:
1909                return (mapping.provider_instance, mapping.item_id)
1910
1911        # As a final fallback, preserve previous behavior.
1912        for mapping in library_item.provider_mappings:
1913            if mapping.provider_instance in user_provider_filter:
1914                return (mapping.provider_instance, mapping.item_id)
1915
1916        # fallback to first mapping
1917        mapping = next(iter(library_item.provider_mappings))
1918        return (mapping.provider_instance, mapping.item_id)
1919
1920    async def _remove_provider_images(self, db_id: int, provider_instance_id: str) -> bool:
1921        """
1922        Remove images belonging to a provider from a library item's stored metadata.
1923
1924        :param db_id: The library (database) id of the item.
1925        :param provider_instance_id: The provider instance whose images should be removed.
1926        :return: True if any images were removed and the db record was updated.
1927        """
1928        # read the raw metadata straight from the db (instead of via get_library_item)
1929        # to avoid persisting any images that are only injected at read time (such as
1930        # the album thumb that gets merged into a track's images)
1931        db_row = await self.mass.music.database.get_row(self.db_table, {"item_id": db_id})
1932        if not db_row or not (raw_metadata := db_row["metadata"]):
1933            return False
1934        metadata = MediaItemMetadata.from_dict(json_loads(raw_metadata))
1935        if not metadata.images:
1936            return False
1937        remaining = UniqueList(
1938            img for img in metadata.images if img.provider != provider_instance_id
1939        )
1940        if len(remaining) == len(metadata.images):
1941            # nothing belonged to this provider
1942            return False
1943        metadata.images = remaining or None
1944        await self.mass.music.database.update(
1945            self.db_table,
1946            {"item_id": db_id},
1947            {"metadata": serialize_to_json(metadata)},
1948        )
1949        return True
1950
1951    def _sync_details_query_parts(self) -> tuple[str, str, dict[str, Any]]:
1952        """
1953        Return extra (columns, joins, params) for this media type's sync-details query.
1954
1955        Override in a subclass to select additional lightweight columns needed by the
1956        library sync change detection for this media type.
1957        """
1958        return "", "", {}
1959
1960    def _parse_sync_details_row(self, db_row: Mapping[str, Any]) -> LibraryItemSyncDetails:
1961        """Parse a raw sync-details db row into a LibraryItemSyncDetails object."""
1962        return LibraryItemSyncDetails(
1963            item_id=db_row["item_id"],
1964            favorite=bool(db_row["favorite"]),
1965            date_added=datetime.fromtimestamp(db_row["timestamp_added"], tz=UTC),
1966            provider_mappings=self._parse_sync_details_mappings(db_row),
1967        )
1968
1969    @final
1970    def _parse_sync_details_mappings(self, db_row: Mapping[str, Any]) -> set[ProviderMapping]:
1971        """Parse the aggregated raw provider mapping rows of a sync-details db row."""
1972        return {
1973            ProviderMapping(
1974                item_id=raw_mapping["item_id"],
1975                provider_domain=raw_mapping["provider_domain"],
1976                provider_instance=raw_mapping["provider_instance"],
1977                available=bool(raw_mapping["available"]),
1978                in_library=parse_optional_bool(raw_mapping["in_library"]),
1979                is_unique=parse_optional_bool(raw_mapping["is_unique"]),
1980            )
1981            for raw_mapping in json_loads(db_row["provider_mappings"])
1982        }
1983
1984    def _parse_summary_row(self, db_row: Mapping[str, Any]) -> MediaItemSummaryType:
1985        """
1986        Parse a raw summary db row into a summary item of this controller's media type.
1987
1988        Override in a subclass to fill additional per-type fields (selected by the
1989        subclass's summary_query).
1990        """
1991        provider_mappings = self._parse_summary_provider_mappings(db_row)
1992        return self.summary_item_cls(
1993            item_id=str(db_row["item_id"]),
1994            provider="library",
1995            name=db_row["name"],
1996            sort_name=db_row["sort_name"],
1997            favorite=bool(db_row["favorite"]),
1998            provider_mappings=provider_mappings,
1999            available=self._summary_available(provider_mappings),
2000            metadata=self._parse_summary_metadata(db_row),
2001        )
2002
2003    @final
2004    @staticmethod
2005    def _parse_summary_provider_mappings(db_row: Mapping[str, Any]) -> set[ProviderMapping]:
2006        """Hydrate the provider mappings of a summary row into ProviderMapping objects."""
2007        if not (raw_mappings := db_row["provider_mappings"]):
2008            return set()
2009        return {ProviderMapping.from_dict(x) for x in json_loads(raw_mappings)}
2010
2011    @final
2012    @staticmethod
2013    def _summary_available(provider_mappings: set[ProviderMapping]) -> bool:
2014        """Compute the availability flag from a summary item's provider mappings."""
2015        # same semantics as the MediaItem.available property
2016        if not (available_providers := get_global_cache_value("available_providers")):
2017            return any(x.available for x in provider_mappings)
2018        if TYPE_CHECKING:
2019            available_providers = cast("set[str]", available_providers)
2020        return any(
2021            x.available and x.provider_instance in available_providers for x in provider_mappings
2022        )
2023
2024    @final
2025    @staticmethod
2026    def _parse_summary_metadata(db_row: Mapping[str, Any]) -> MediaItemMetadataSummary:
2027        """Build the slim metadata of a summary row, carrying only the (first) thumb image."""
2028        thumb: MediaItemImage | None = None
2029        if raw_images := db_row["images"]:
2030            for image in json_loads(raw_images):
2031                if image["type"] != ImageType.THUMB.value:
2032                    continue
2033                thumb = MediaItemImage(
2034                    type=ImageType.THUMB,
2035                    path=image["path"],
2036                    provider=image["provider"],
2037                    remotely_accessible=image.get("remotely_accessible", False),
2038                )
2039                break
2040        return MediaItemMetadataSummary(images=UniqueList([thumb]) if thumb else None)
2041
2042    @final
2043    def _parse_summary_artist_mappings(
2044        self, db_row: Mapping[str, Any]
2045    ) -> UniqueList[ItemMappingSummary]:
2046        """Parse the aggregated slim artist mapping rows of a summary db row."""
2047        return UniqueList(
2048            ItemMappingSummary(
2049                media_type=MediaType.ARTIST,
2050                item_id=str(raw_mapping["item_id"]),
2051                provider="library",
2052                name=raw_mapping["name"],
2053                sort_name=raw_mapping["sort_name"],
2054            )
2055            for raw_mapping in json_loads(db_row["artists"])
2056        )
2057
2058    async def _adapt_query_for_collections(
2059        self,
2060        sql_query: str,
2061        query_params: dict[str, Any],
2062        summary: bool,
2063        order_by: str | None,
2064        collection_name: str | None = None,
2065        search: str | None = None,
2066    ) -> str:
2067        cache_key_json_object = f"collection_{self.api_base}"
2068        json_object = await self.mass.cache.get(key=cache_key_json_object, category=int(summary))
2069        if json_object is None:
2070            # get column names of base query
2071            db_rows = await self.mass.music.database.get_rows_from_query(
2072                sql_query, query_params, limit=1, offset=0
2073            )
2074            # create a sql json_object which queries all these columns
2075            if db_rows:
2076                json_object = (
2077                    "json_object(" + ",".join([f"'{x}',{x}" for x in db_rows[0].keys()]) + ")"  # noqa: SIM118
2078                )
2079                await self.mass.cache.set(
2080                    key=cache_key_json_object, category=int(summary), data=json_object
2081                )
2082            else:
2083                json_object = "json_object()"
2084
2085        collections_column = "collections" if summary else "json_extract(metadata, '$.collections')"
2086
2087        supported_order_keys = [
2088            "name",
2089            "name_desc",
2090            "sort_name",
2091            "sort_name_desc",
2092            "timestamp_added",
2093            "timestamp_added_desc",
2094            "timestamp_modified",
2095            "timestamp_modified_desc",
2096            "last_played",
2097            "last_played_desc",
2098            "play_count",
2099            "play_count_desc",
2100        ]
2101
2102        # additional order options subject to media type
2103        # single is targeting a single media item, collection the aggregated ones
2104        single_extra_order_keys = ""
2105        collection_extra_order_keys = ""
2106        if MediaType.AUDIOBOOK.value in self.api_base:
2107            single_extra_order_keys = "duration,"
2108            collection_extra_order_keys = "SUM(duration) as duration,"
2109            supported_order_keys += ["duration", "duration_desc"]
2110
2111        sql_query = f"""
2112        SELECT * FROM (
2113
2114            WITH
2115                joined_table as ({sql_query}),
2116                collection_extract as (
2117                    SELECT
2118                        name as media_name,
2119                        timestamp_added,
2120                        timestamp_modified,
2121                        last_played,
2122                        play_count,
2123                        {single_extra_order_keys}
2124                        json_extract(iter_coll.value, '$.title') as collection_title,
2125                        json_extract(iter_coll.value, '$.sequence') as collection_sequence,
2126                        json_extract(iter_coll.value, '$.search_title') as collection_search_title,
2127                        json_extract(iter_coll.value, '$.search_sort_title') as collection_search_sort_title,
2128                        CASE
2129                            WHEN json_type(iter_coll.value, '$.sequence') IN ('integer', 'real')
2130                            THEN 1
2131                            WHEN json_type(iter_coll.value, '$.sequence') = 'text'
2132                                AND json_valid(json_extract(iter_coll.value, '$.sequence'))
2133                            THEN CASE
2134                                WHEN json_type(json_extract(iter_coll.value, '$.sequence'))
2135                                    IN ('integer', 'real')
2136                                THEN 1
2137                                ELSE 0
2138                            END
2139                            ELSE 0
2140                        END as collection_sequence_is_numeric,
2141                        {json_object} as media_data
2142                    FROM (
2143                        SELECT * FROM joined_table
2144                    ), json_each({collections_column}) as iter_coll
2145                )
2146            SELECT
2147                'collection' as type,
2148                collection_title as name,
2149                COALESCE(MAX(collection_search_title), replace(lower(collection_title),' ','')) AS search_name,
2150                COALESCE(MAX(collection_search_sort_title), replace(lower(collection_title),' ','')) AS search_sort_name,
2151                MAX(timestamp_added) as timestamp_added,
2152                MAX(timestamp_modified) as timestamp_modified,
2153                MAX(last_played) as last_played,
2154                SUM(play_count) as play_count,
2155                {collection_extra_order_keys}
2156                json_group_array(media_data) as media_data
2157            FROM (
2158                SELECT * FROM collection_extract
2159                -- NOTE: The following ORDER_BY to control the aggregation order of json_group_array is undocumented sqlite behavior
2160                -- Confirmed working with sqlite 3.40.1 & 3.53
2161                -- Once our image moves to sqlite 3.44 we can and should make use of ORDER_BY in the aggregate itself
2162                ORDER BY collection_title,
2163                -- null case
2164                CASE WHEN collection_sequence IS NULL THEN 1 ELSE 0 END,
2165                -- numeric before text
2166                CASE WHEN collection_sequence_is_numeric THEN 0 ELSE 1 END,
2167                -- order NUMERIC
2168                CASE WHEN collection_sequence_is_numeric
2169                    THEN CAST(collection_sequence AS REAL)
2170                END,
2171                -- order TEXT
2172                CASE WHEN NOT collection_sequence_is_numeric
2173                    THEN collection_sequence
2174                END COLLATE NOCASE,
2175                -- order by media name if no sequence given
2176                CASE
2177                    WHEN collection_sequence IS NULL
2178                    THEN media_name
2179                END COLLATE NOCASE
2180            )
2181            GROUP BY collection_title
2182
2183            UNION ALL
2184
2185            SELECT 'single', name, search_name, search_sort_name,
2186                timestamp_added, timestamp_modified, last_played, play_count,
2187                {single_extra_order_keys}
2188                {json_object} FROM joined_table
2189                WHERE {collections_column} IS NULL
2190                    OR {collections_column} = '[]'
2191        )
2192        """
2193
2194        if collection_name:
2195            sql_query += " WHERE type = 'collection' AND name = :collection_name"
2196            return sql_query
2197
2198        if search:
2199            sql_query += " WHERE search_name LIKE :search"
2200
2201        if order_by:
2202            if order_by not in supported_order_keys:
2203                self.logger.warning("%s is not supported for order_by key in collections", order_by)
2204                order_by = "name"  # fallback
2205            if sort_key := SORT_KEYS.get(order_by):
2206                sql_query += f" ORDER BY {sort_key}"
2207
2208        return sql_query
2209