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        """Update the provider_items table for the media item."""
1208        db_id = int(item_id)  # ensure integer
1209        if overwrite:
1210            # on overwrite, clear the provider_mappings table first
1211            # this is done for filesystem provider changing the path (and thus item_id)
1212            await self.mass.music.database.delete(
1213                DB_TABLE_PROVIDER_MAPPINGS,
1214                {"media_type": self.media_type.value, "item_id": db_id},
1215            )
1216        prov_map_objs: list[dict[str, Any]] = []
1217        for provider_mapping in provider_mappings:
1218            prov_map_obj = {
1219                "media_type": self.media_type.value,
1220                "item_id": db_id,
1221                "provider_domain": provider_mapping.provider_domain,
1222                "provider_instance": provider_mapping.provider_instance,
1223                "provider_item_id": provider_mapping.item_id,
1224                "available": provider_mapping.available,
1225                "audio_format": serialize_to_json(provider_mapping.audio_format),
1226            }
1227            for key in ("url", "details", "in_library", "is_unique"):
1228                if (value := getattr(provider_mapping, key, None)) is not None:
1229                    prov_map_obj[key] = value
1230            prov_map_objs.append(prov_map_obj)
1231        await self.mass.music.database.upsert_many(
1232            DB_TABLE_PROVIDER_MAPPINGS,
1233            prov_map_objs,
1234        )
1235
1236    @final
1237    async def set_external_ids(
1238        self,
1239        item_id: str | int,
1240        external_ids: Iterable[tuple[ExternalID, str]],
1241    ) -> None:
1242        """Update the external_id_lookup table rows for the media item."""
1243        db_id = int(item_id)  # ensure integer
1244        await self.mass.music.database.delete(
1245            DB_TABLE_EXTERNAL_ID_LOOKUP,
1246            {"media_type": self.media_type.value, "item_id": db_id},
1247        )
1248        if lookup_rows := [
1249            {
1250                "media_type": self.media_type.value,
1251                "external_id_type": external_id_type,
1252                "external_id": external_id,
1253                "item_id": db_id,
1254            }
1255            for external_id_type, external_id in external_ids
1256        ]:
1257            await self.mass.music.database.upsert_many(DB_TABLE_EXTERNAL_ID_LOOKUP, lookup_rows)
1258
1259    @abstractmethod
1260    async def match_providers(self, db_item: ItemCls) -> None:
1261        """
1262        Try to find match on all (streaming) providers for the provided (database) item.
1263
1264        This is used to link objects of different providers/qualities together.
1265        """
1266
1267    if TYPE_CHECKING:
1268
1269        @overload
1270        async def get_library_items_by_query(
1271            self,
1272            favorite: bool | None = None,
1273            search: str | None = None,
1274            limit: int = 500,
1275            offset: int = 0,
1276            order_by: str | None = None,
1277            provider_filter: list[str] | None = None,
1278            extra_query_parts: list[str] | None = None,
1279            extra_query_params: dict[str, Any] | None = None,
1280            extra_join_parts: list[str] | None = None,
1281            genre_ids: int | list[int] | None = None,
1282            played_only: bool = False,
1283            in_library_only: bool = False,
1284            summary: bool = False,
1285            *,
1286            collapse_collections: Literal[True],
1287        ) -> list[ItemCls | MediaCollection[ItemCls]]: ...
1288
1289        @overload
1290        async def get_library_items_by_query(
1291            self,
1292            favorite: bool | None = None,
1293            search: str | None = None,
1294            limit: int = 500,
1295            offset: int = 0,
1296            order_by: str | None = None,
1297            provider_filter: list[str] | None = None,
1298            extra_query_parts: list[str] | None = None,
1299            extra_query_params: dict[str, Any] | None = None,
1300            extra_join_parts: list[str] | None = None,
1301            genre_ids: int | list[int] | None = None,
1302            played_only: bool = False,
1303            in_library_only: bool = False,
1304            summary: bool = False,
1305            *,
1306            collapse_collections: Literal[False] = False,
1307        ) -> list[ItemCls]: ...
1308
1309        @overload
1310        async def get_library_items_by_query(
1311            self,
1312            favorite: bool | None = None,
1313            search: str | None = None,
1314            limit: int = 500,
1315            offset: int = 0,
1316            order_by: str | None = None,
1317            provider_filter: list[str] | None = None,
1318            extra_query_parts: list[str] | None = None,
1319            extra_query_params: dict[str, Any] | None = None,
1320            extra_join_parts: list[str] | None = None,
1321            genre_ids: int | list[int] | None = None,
1322            played_only: bool = False,
1323            in_library_only: bool = False,
1324            summary: bool = False,
1325            *,
1326            collapse_collections: bool,
1327        ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]: ...
1328
1329    @final
1330    async def get_library_items_by_query(  # noqa: PLR0913
1331        self,
1332        favorite: bool | None = None,
1333        search: str | None = None,
1334        limit: int = 500,
1335        offset: int = 0,
1336        order_by: str | None = None,
1337        provider_filter: list[str] | None = None,
1338        extra_query_parts: list[str] | None = None,
1339        extra_query_params: dict[str, Any] | None = None,
1340        extra_join_parts: list[str] | None = None,
1341        genre_ids: int | list[int] | None = None,
1342        played_only: bool = False,
1343        in_library_only: bool = False,
1344        summary: bool = False,
1345        *,
1346        collapse_collections: bool = False,
1347    ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]:
1348        """Fetch MediaItem records from database by building the query."""
1349        query_params = dict(extra_query_params) if extra_query_params else {}
1350        query_parts: list[str] = list(extra_query_parts) if extra_query_parts else []
1351        join_parts: list[str] = list(extra_join_parts) if extra_join_parts else []
1352        search = self._preprocess_search(search)
1353        genre_ids = self._preprocess_genre_ids(genre_ids)
1354        # create special performant random query
1355        if order_by and order_by.startswith("random"):
1356            self._apply_random_subquery(
1357                query_parts=query_parts,
1358                query_params=query_params,
1359                join_parts=join_parts,
1360                favorite=favorite,
1361                search=search if not collapse_collections else None,
1362                genre_ids=genre_ids,
1363                provider_filter=provider_filter,
1364                played_only=played_only,
1365                limit=limit,
1366                in_library_only=in_library_only,
1367            )
1368        else:
1369            # apply filters
1370            self._apply_filters(
1371                query_parts=query_parts,
1372                query_params=query_params,
1373                favorite=favorite,
1374                search=search if not collapse_collections else None,
1375                genre_ids=genre_ids,
1376                provider_filter=provider_filter,
1377                played_only=played_only,
1378                in_library_only=in_library_only,
1379            )
1380        # build and execute final query
1381        sql_query, base_query_params = self._build_final_query(
1382            query_parts, join_parts, order_by, summary=summary
1383        )
1384        # base query params act as defaults: callers may override them via extra_query_params
1385        for key, value in base_query_params.items():
1386            query_params.setdefault(key, value)
1387
1388        if collapse_collections:
1389            if search:
1390                query_params["search"] = f"%{search}%"
1391            sql_query = await self._adapt_query_for_collections(
1392                sql_query, query_params, summary=summary, order_by=order_by, search=search
1393            )
1394
1395        db_rows = await self.mass.music.database.get_rows_from_query(
1396            sql_query, query_params, limit=limit, offset=offset
1397        )
1398        if collapse_collections:
1399            items: list[ItemCls | MediaCollection[ItemCls]] = []
1400
1401            def _parse_method(x: str) -> ItemCls:
1402                if summary:
1403                    return cast("ItemCls", self._parse_summary_row(json_loads(x)))
1404                return cast(
1405                    "ItemCls",
1406                    self.item_cls.from_dict(self._parse_db_row(json_loads(x))),
1407                )
1408
1409            for db_row in db_rows:
1410                if db_row["type"] == "single":
1411                    items.append(_parse_method(db_row["media_data"]))
1412                elif db_row["type"] == "collection":
1413                    items.append(
1414                        MediaCollection[ItemCls](
1415                            item_id=get_collection_item_id(
1416                                db_row["name"], item_media_type=self.media_type
1417                            ),
1418                            name=db_row["name"],
1419                            provider="library",
1420                            provider_mappings=set(),
1421                            items=UniqueList(
1422                                [_parse_method(x) for x in json_loads(db_row["media_data"])]
1423                            ),
1424                        )
1425                    )
1426            return items
1427        if summary:
1428            return [cast("ItemCls", self._parse_summary_row(db_row)) for db_row in db_rows]
1429        return [
1430            cast("ItemCls", self.item_cls.from_dict(self._parse_db_row(db_row)))
1431            for db_row in db_rows
1432        ]
1433
1434    @final
1435    async def _get_library_item_by_match(self, item: ItemCls | ItemMapping) -> int | None:
1436        if item.provider == "library":
1437            return int(item.item_id)
1438        # search by provider mappings if item is ItemMapping
1439        if isinstance(item, ItemMapping):
1440            if cur_item := await self.get_library_item_by_prov_id(item.item_id, item.provider):
1441                return int(cur_item.item_id)
1442
1443        # for all other items that are MediaItemType, check provider_mappings if it exists
1444        provider_mappings = getattr(item, "provider_mappings", None)
1445        if provider_mappings:
1446            if cur_item := await self.get_library_item_by_prov_mappings(provider_mappings):
1447                return int(cur_item.item_id)
1448        if cur_item := await self.get_library_item_by_external_ids(item.external_ids):
1449            # existing item match by external id
1450            # Double check external IDs - if MBID exists, regards that as overriding
1451            if compare_media_item(item, cur_item):
1452                return int(cur_item.item_id)
1453        # search by (exact) name match
1454        query = f"{self.db_table}.name = :name OR {self.db_table}.sort_name = :sort_name"
1455        query_params = {"name": item.name, "sort_name": item.sort_name}
1456        for db_item in await self.get_library_items_by_query(
1457            extra_query_parts=[query], extra_query_params=query_params
1458        ):
1459            if compare_media_item(db_item, item, True):
1460                return int(db_item.item_id)
1461        return None
1462
1463    def _external_ids_query(
1464        self, media_type: MediaType | None = None, table_alias: str | None = None
1465    ) -> str:
1466        """
1467        Return a subquery that selects the external ids of a media item as a JSON array.
1468
1469        :param media_type: Media type to select the external ids for, defaults to
1470            this controller's media type.
1471        :param table_alias: (Aliased) table name the subquery correlates against,
1472            defaults to this controller's table.
1473        """
1474        media_type = media_type or self.media_type
1475        table_alias = table_alias or self.db_table
1476        return (
1477            f"(SELECT JSON_GROUP_ARRAY(json_array("
1478            f"{DB_TABLE_EXTERNAL_ID_LOOKUP}.external_id_type, "
1479            f"{DB_TABLE_EXTERNAL_ID_LOOKUP}.external_id)) "
1480            f"FROM {DB_TABLE_EXTERNAL_ID_LOOKUP} "
1481            f"WHERE {DB_TABLE_EXTERNAL_ID_LOOKUP}.media_type = '{media_type.value}' "
1482            f"AND {DB_TABLE_EXTERNAL_ID_LOOKUP}.item_id = {table_alias}.item_id)"
1483        )
1484
1485    def _provider_mappings_query(self) -> str:
1486        """Return a subquery that selects the provider mappings of a media item as a JSON array."""
1487        return f"""(SELECT JSON_GROUP_ARRAY(
1488            json_object(
1489                'item_id', pm.provider_item_id,
1490                'provider_domain', pm.provider_domain,
1491                'provider_instance', pm.provider_instance,
1492                'available', pm.available,
1493                'audio_format', json(pm.audio_format),
1494                'url', pm.url,
1495                'details', pm.details,
1496                'in_library', pm.in_library,
1497                'is_unique', pm.is_unique
1498            )) FROM {DB_TABLE_PROVIDER_MAPPINGS} pm
1499            WHERE pm.item_id = {self.db_table}.item_id
1500            AND pm.media_type = '{self.media_type.value}')"""
1501
1502    def _artist_mappings_summary_query(
1503        self, m2m_table: str, m2m_key: str, include_artist_type: bool = False
1504    ) -> str:
1505        """
1506        Return a subquery selecting the slim artist mappings JSON of a summary row.
1507
1508        :param m2m_table: The many-to-many table linking artists to this media type.
1509        :param m2m_key: The column in the m2m table referencing this media type's item id.
1510        :param include_artist_type: Also select the artist_type of each artist.
1511        """
1512        artist_type_part = ",\n                'artist_type', artists.artist_type"
1513        return f"""(SELECT JSON_GROUP_ARRAY(
1514            json_object(
1515                'item_id', artists.item_id,
1516                'name', artists.name,
1517                'sort_name', artists.sort_name{artist_type_part if include_artist_type else ""}
1518            )) FROM artists
1519            JOIN {m2m_table} ON artists.item_id = {m2m_table}.artist_id
1520            WHERE {m2m_table}.{m2m_key} = {self.db_table}.item_id)"""
1521
1522    def _summary_base_columns(self) -> str:
1523        """Return the SELECT columns shared by every summary query."""
1524        # the search/sort/statistics columns are selected so ORDER BY (see SORT_KEYS)
1525        # resolves them from the result set, like the full query's SELECT * does
1526        return f"""
1527            {self.db_table}.item_id,
1528            {self.db_table}.name,
1529            {self.db_table}.sort_name,
1530            {self.db_table}.favorite,
1531            {self.db_table}.search_name AS search_name,
1532            {self.db_table}.search_sort_name AS search_sort_name,
1533            {self.db_table}.play_count AS play_count,
1534            {self.db_table}.last_played AS last_played,
1535            {self.db_table}.timestamp_added AS timestamp_added,
1536            {self.db_table}.timestamp_modified AS timestamp_modified,
1537            json_extract({self.db_table}.metadata, '$.images') AS images,
1538            json_extract({self.db_table}.metadata, '$.collections') AS collections"""
1539
1540    async def _localized_search_fallback(
1541        self, search_query: str, limit: int, offset: int = 0, **call_kwargs: Any
1542    ) -> list[ItemCls]:
1543        """
1544        Retry a library search using the canonical names behind a localized query.
1545
1546        For genre/playlist searches that return nothing literally, reverse-resolve the query to the
1547        canonical (English) names of matching localized items and search those, so an item is
1548        findable by the localized name the user sees. The caller's other filters (favorite,
1549        order_by, provider and any controller-specific kwargs) are forwarded unchanged so the retry
1550        behaves like the literal search; results are merged, de-duplicated and paginated here. See
1551        ``TranslationController.reverse_lookup_media_names``.
1552        """
1553        seen: set[Any] = set()
1554        merged: list[ItemCls] = []
1555        # iterate the canonical names in a stable order, and fetch each from the start so the
1556        # offset/limit window can be applied to the merged, de-duplicated result set
1557        for name in sorted(await self.mass.translations.reverse_lookup_media_names(search_query)):
1558            for item in await self.library_items(
1559                search=name,
1560                limit=limit + offset,
1561                offset=0,
1562                _localized_fallback=False,
1563                **call_kwargs,
1564            ):
1565                if item.item_id not in seen:
1566                    seen.add(item.item_id)
1567                    merged.append(item)
1568        return merged[offset : offset + limit]
1569
1570    @abstractmethod
1571    async def _add_library_item(
1572        self,
1573        item: ItemCls,
1574        overwrite_existing: bool = False,
1575    ) -> int:
1576        """Add item to library and return the database id."""
1577
1578    @abstractmethod
1579    async def _update_library_item(
1580        self, item_id: str | int, update: ItemCls, overwrite: bool = False
1581    ) -> None:
1582        """Update existing library record in the database."""
1583
1584    def _search_filter_clause(self, search: str, query_params: dict[str, Any]) -> str:
1585        """Return the SQL WHERE clause fragment used for search filtering."""
1586        return search_name_match_clause(self.db_table, search, "search", query_params)
1587
1588    @final
1589    def _preprocess_search(self, search: str | None) -> str | None:
1590        """Normalize the search string for use in the search filter clauses."""
1591        return create_safe_string(search, True, True) if search else search
1592
1593    @final
1594    @staticmethod
1595    def _preprocess_genre_ids(genre_ids: int | list[int] | None) -> list[int] | None:
1596        if genre_ids is None:
1597            return None
1598        if isinstance(genre_ids, list):
1599            normalized = [int(x) for x in genre_ids]
1600        else:
1601            normalized = [int(genre_ids)]
1602        return normalized or None
1603
1604    @final
1605    @staticmethod
1606    def _clean_query_parts(query_parts: list[str]) -> list[str]:
1607        """Clean the query parts list by removing duplicate where statements."""
1608        return [x[5:] if x.lower().startswith("where ") else x for x in query_parts]
1609
1610    @final
1611    def _apply_random_subquery(
1612        self,
1613        query_parts: list[str],
1614        query_params: dict[str, Any],
1615        join_parts: list[str],
1616        favorite: bool | None,
1617        search: str | None,
1618        genre_ids: list[int] | None,
1619        provider_filter: list[str] | None,
1620        played_only: bool = False,
1621        limit: int = 500,
1622        in_library_only: bool = False,
1623    ) -> None:
1624        """Build a fast random subquery with all filters applied."""
1625        sub_query_parts = query_parts.copy()
1626        sub_join_parts = join_parts.copy()
1627
1628        # Apply all filters to the subquery
1629        self._apply_filters(
1630            query_parts=sub_query_parts,
1631            query_params=query_params,
1632            favorite=favorite,
1633            search=search,
1634            genre_ids=genre_ids,
1635            provider_filter=provider_filter,
1636            played_only=played_only,
1637            in_library_only=in_library_only,
1638        )
1639
1640        # Build the subquery
1641        sub_query = f"SELECT {self.db_table}.item_id FROM {self.db_table}"
1642
1643        if sub_join_parts:
1644            sub_query += f" {' '.join(sub_join_parts)}"
1645
1646        if sub_query_parts:
1647            sub_query += " WHERE " + " AND ".join(self._clean_query_parts(sub_query_parts))
1648
1649        sub_query += f" ORDER BY RANDOM() LIMIT {limit}"
1650
1651        # The query now only consists of the random subquery, which applies all filters
1652        # within itself
1653        query_parts.clear()
1654        query_parts.append(f"{self.db_table}.item_id in ({sub_query})")
1655        join_parts.clear()
1656
1657    @final
1658    def _apply_filters(
1659        self,
1660        query_parts: list[str],
1661        query_params: dict[str, Any],
1662        favorite: bool | None,
1663        search: str | None,
1664        genre_ids: list[int] | None,
1665        provider_filter: list[str] | None,
1666        played_only: bool = False,
1667        in_library_only: bool = False,
1668    ) -> None:
1669        """Apply search, favorite, and provider filters."""
1670        # handle search
1671        if search:
1672            query_parts.append(self._search_filter_clause(search, query_params))
1673        # handle favorite filter
1674        if favorite is not None:
1675            query_parts.append(f"{self.db_table}.favorite = :favorite")
1676            query_params["favorite"] = favorite
1677        # handle played_only filter
1678        if played_only:
1679            query_parts.append(f"{self.db_table}.last_played > 0")
1680        # handle genre filter
1681        if genre_ids:
1682            query_params["genre_ids"] = genre_ids
1683            query_params["genre_media_type"] = self.media_type.value
1684            query_parts.append(
1685                f"EXISTS("
1686                f"SELECT 1 FROM {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING} gm "
1687                f"WHERE gm.media_id = {self.db_table}.item_id "
1688                "AND gm.media_type = :genre_media_type "
1689                "AND gm.genre_id IN :genre_ids)"
1690            )
1691        # Apply the provider filter
1692        if provider_filter or in_library_only:
1693            query_parts.append(
1694                self._provider_filter_clause(query_params, provider_filter, in_library_only)
1695            )
1696
1697    @final
1698    def _provider_filter_clause(
1699        self,
1700        query_params: dict[str, Any],
1701        provider_filter: list[str] | None,
1702        in_library_only: bool = False,
1703    ) -> str:
1704        """
1705        Return the SQL clause that restricts items by their provider mappings.
1706
1707        At least one of provider_filter/in_library_only must be set, otherwise the
1708        returned clause only asserts that the item has any mapping at all.
1709
1710        :param query_params: Query params dict; the clause's bound params are added to it.
1711        :param provider_filter: Only match items mapped to one of these provider instances.
1712        :param in_library_only: Only match provider mappings that are in the provider's library.
1713        """
1714        # NOTE: provider mapping filters are applied as a correlated EXISTS subquery
1715        # instead of a JOIN + GROUP BY, so SQLite can stream results straight from the
1716        # sort index instead of materializing/sorting the whole (deduped) result set.
1717        query_params["provider_media_type"] = self.media_type.value
1718        conditions = [
1719            f"provider_mappings.item_id = {self.db_table}.item_id",
1720            "provider_mappings.media_type = :provider_media_type",
1721        ]
1722        if in_library_only:
1723            conditions.append("provider_mappings.in_library = 1")
1724        if provider_filter:
1725            provider_conditions = []
1726            for idx, prov in enumerate(provider_filter):
1727                param_name = f"provider_filter_{idx}"
1728                provider_conditions.append(f"provider_mappings.provider_instance = :{param_name}")
1729                query_params[param_name] = prov
1730            conditions.append(f"({' OR '.join(provider_conditions)})")
1731        return f"EXISTS(SELECT 1 FROM provider_mappings WHERE {' AND '.join(conditions)})"
1732
1733    @final
1734    def _build_final_query(
1735        self,
1736        query_parts: list[str],
1737        join_parts: list[str],
1738        order_by: str | None,
1739        summary: bool = False,
1740    ) -> tuple[str, dict[str, Any]]:
1741        """Build the final SQL query string and its (base) bound query params."""
1742        sql_query, base_query_params = self.summary_query if summary else self.base_query
1743
1744        # Add joins
1745        if join_parts:
1746            sql_query += f" {' '.join(join_parts)} "
1747
1748        # Add where clauses
1749        if query_parts:
1750            # prevent duplicate where statement
1751            sql_query += " WHERE " + " AND ".join(self._clean_query_parts(query_parts))
1752
1753        # Add grouping (only needed when caller-provided joins can fan out rows)
1754        # and ordering. Without a GROUP BY, SQLite can stream results directly
1755        # from the sort index instead of sorting the whole result set.
1756        if join_parts:
1757            sql_query += f" GROUP BY {self.db_table}.item_id"
1758
1759        if order_by:
1760            if sort_key := SORT_KEYS.get(order_by):
1761                sql_query += f" ORDER BY {sort_key}"
1762
1763        return sql_query, base_query_params
1764
1765    @final
1766    @staticmethod
1767    def _parse_db_row(db_row: Mapping[str, Any]) -> dict[str, Any]:
1768        """Parse raw db Mapping into a dict."""
1769        db_row_dict = dict(db_row)
1770        db_row_dict["provider"] = "library"
1771        db_row_dict["favorite"] = bool(db_row_dict["favorite"])
1772        db_row_dict["item_id"] = str(db_row_dict["item_id"])
1773        db_row_dict["date_added"] = datetime.fromtimestamp(
1774            db_row_dict["timestamp_added"], tz=UTC
1775        ).isoformat()
1776
1777        for key in JSON_KEYS:
1778            if key not in db_row_dict:
1779                continue
1780            if not (raw_value := db_row_dict[key]):
1781                continue
1782            db_row_dict[key] = json_loads(raw_value)
1783
1784        # parse "fully_played" as bool if present in the row
1785        if "fully_played" in db_row_dict:
1786            db_row_dict["fully_played"] = parse_optional_bool(db_row_dict["fully_played"])
1787
1788        # copy track_album --> album
1789        if track_album := db_row_dict.get("track_album"):
1790            db_row_dict["album"] = track_album
1791            db_row_dict["disc_number"] = track_album["disc_number"]
1792            db_row_dict["track_number"] = track_album["track_number"]
1793            # always prefer album image over track image
1794            if (album_images := track_album.get("images")) and (
1795                album_thumb := next((x for x in album_images if x["type"] == "thumb"), None)
1796            ):
1797                # copy album image to itemmapping single image (on the track)
1798                db_row_dict["image"] = album_thumb
1799                # also set image on the album dict for ItemMapping compatibility
1800                track_album["image"] = album_thumb
1801                if db_row_dict["metadata"].get("images"):
1802                    # merge album image with existing images
1803                    db_row_dict["metadata"]["images"] = [
1804                        album_thumb,
1805                        *db_row_dict["metadata"]["images"],
1806                    ]
1807                else:
1808                    db_row_dict["metadata"]["images"] = [album_thumb]
1809
1810        if audiobook_artists := db_row_dict.get("audiobook_artists"):
1811            _narrators = []
1812            _authors = []
1813            for artist in audiobook_artists:
1814                artist_type = artist.get("artist_type")
1815                if artist_type == "author":
1816                    _authors.append(artist)
1817                elif artist_type == "narrator":
1818                    _narrators.append(artist)
1819            if _authors:
1820                # prevent overwriting string values
1821                db_row_dict["authors"] = _authors
1822            if _narrators:
1823                # prevent overwriting string values
1824                db_row_dict["narrators"] = _narrators
1825
1826        return db_row_dict
1827
1828    @final
1829    def _ensure_provider_filter(
1830        self,
1831        provider: str | list[str] | None,
1832    ) -> list[str] | None:
1833        """Ensure the provider filter respects the current user's provider filter."""
1834        # Apply user provider filter if needed
1835        user = get_current_user()
1836        user_provider_filter = user.provider_filter if user and user.provider_filter else None
1837        final_provider_filter: list[str] | None = None
1838        if user_provider_filter:
1839            plugin_provider_instances = {
1840                prov.instance_id for prov in self.mass.providers if prov.type == ProviderType.PLUGIN
1841            }
1842            # User has a provider filter set
1843            if provider:
1844                # Explicit provider filter provided - validate against user's allowed providers
1845                requested_providers = [provider] if isinstance(provider, str) else provider
1846                # Only restrict access to music providers.
1847                final_provider_filter = [
1848                    p
1849                    for p in requested_providers
1850                    if p in user_provider_filter or p in plugin_provider_instances
1851                ]
1852                if not final_provider_filter:
1853                    # No overlap - user requested providers they don't have access to
1854                    raise InsufficientPermissions(
1855                        "User does not have permission to access the requested provider(s)."
1856                    )
1857            else:
1858                # No explicit filter - apply user music provider filter but keep plugin providers.
1859                final_provider_filter = list(
1860                    dict.fromkeys([*user_provider_filter, *plugin_provider_instances])
1861                )
1862        elif provider is not None:
1863            # No user filter - use the provided filter as is
1864            final_provider_filter = [provider] if isinstance(provider, str) else provider
1865        return final_provider_filter
1866
1867    @final
1868    def _select_provider_id(self, library_item: ItemCls) -> tuple[str, str]:
1869        """Select the correct provider id to use for fetching the item."""
1870        if not library_item.provider_mappings:
1871            msg = (
1872                f"{self.media_type.value} {library_item.item_id} "
1873                "is no longer available on any provider"
1874            )
1875            raise MediaNotFoundError(msg)
1876        user = get_current_user()
1877        user_provider_filter = user.provider_filter if user and user.provider_filter else None
1878        if not user_provider_filter:
1879            mapping = next(iter(library_item.provider_mappings))
1880            return (mapping.provider_instance, mapping.item_id)
1881
1882        # First prefer music provider mappings that are explicitly allowed for this user.
1883        # prefer user provider filter if available
1884        for mapping in library_item.provider_mappings:
1885            provider = self.mass.get_provider(mapping.provider_instance)
1886            if provider and provider.type == ProviderType.MUSIC:
1887                if mapping.provider_instance in user_provider_filter:
1888                    return (mapping.provider_instance, mapping.item_id)
1889
1890        # If no allowed music mapping exists, fall back to plugin mappings.
1891        for mapping in library_item.provider_mappings:
1892            provider = self.mass.get_provider(mapping.provider_instance)
1893            if provider and provider.type == ProviderType.PLUGIN:
1894                return (mapping.provider_instance, mapping.item_id)
1895
1896        # As a final fallback, preserve previous behavior.
1897        for mapping in library_item.provider_mappings:
1898            if mapping.provider_instance in user_provider_filter:
1899                return (mapping.provider_instance, mapping.item_id)
1900
1901        # fallback to first mapping
1902        mapping = next(iter(library_item.provider_mappings))
1903        return (mapping.provider_instance, mapping.item_id)
1904
1905    async def _remove_provider_images(self, db_id: int, provider_instance_id: str) -> bool:
1906        """
1907        Remove images belonging to a provider from a library item's stored metadata.
1908
1909        :param db_id: The library (database) id of the item.
1910        :param provider_instance_id: The provider instance whose images should be removed.
1911        :return: True if any images were removed and the db record was updated.
1912        """
1913        # read the raw metadata straight from the db (instead of via get_library_item)
1914        # to avoid persisting any images that are only injected at read time (such as
1915        # the album thumb that gets merged into a track's images)
1916        db_row = await self.mass.music.database.get_row(self.db_table, {"item_id": db_id})
1917        if not db_row or not (raw_metadata := db_row["metadata"]):
1918            return False
1919        metadata = MediaItemMetadata.from_dict(json_loads(raw_metadata))
1920        if not metadata.images:
1921            return False
1922        remaining = UniqueList(
1923            img for img in metadata.images if img.provider != provider_instance_id
1924        )
1925        if len(remaining) == len(metadata.images):
1926            # nothing belonged to this provider
1927            return False
1928        metadata.images = remaining or None
1929        await self.mass.music.database.update(
1930            self.db_table,
1931            {"item_id": db_id},
1932            {"metadata": serialize_to_json(metadata)},
1933        )
1934        return True
1935
1936    def _sync_details_query_parts(self) -> tuple[str, str, dict[str, Any]]:
1937        """
1938        Return extra (columns, joins, params) for this media type's sync-details query.
1939
1940        Override in a subclass to select additional lightweight columns needed by the
1941        library sync change detection for this media type.
1942        """
1943        return "", "", {}
1944
1945    def _parse_sync_details_row(self, db_row: Mapping[str, Any]) -> LibraryItemSyncDetails:
1946        """Parse a raw sync-details db row into a LibraryItemSyncDetails object."""
1947        return LibraryItemSyncDetails(
1948            item_id=db_row["item_id"],
1949            favorite=bool(db_row["favorite"]),
1950            date_added=datetime.fromtimestamp(db_row["timestamp_added"], tz=UTC),
1951            provider_mappings=self._parse_sync_details_mappings(db_row),
1952        )
1953
1954    @final
1955    def _parse_sync_details_mappings(self, db_row: Mapping[str, Any]) -> set[ProviderMapping]:
1956        """Parse the aggregated raw provider mapping rows of a sync-details db row."""
1957        return {
1958            ProviderMapping(
1959                item_id=raw_mapping["item_id"],
1960                provider_domain=raw_mapping["provider_domain"],
1961                provider_instance=raw_mapping["provider_instance"],
1962                available=bool(raw_mapping["available"]),
1963                in_library=parse_optional_bool(raw_mapping["in_library"]),
1964                is_unique=parse_optional_bool(raw_mapping["is_unique"]),
1965            )
1966            for raw_mapping in json_loads(db_row["provider_mappings"])
1967        }
1968
1969    def _parse_summary_row(self, db_row: Mapping[str, Any]) -> MediaItemSummaryType:
1970        """
1971        Parse a raw summary db row into a summary item of this controller's media type.
1972
1973        Override in a subclass to fill additional per-type fields (selected by the
1974        subclass's summary_query).
1975        """
1976        provider_mappings = self._parse_summary_provider_mappings(db_row)
1977        return self.summary_item_cls(
1978            item_id=str(db_row["item_id"]),
1979            provider="library",
1980            name=db_row["name"],
1981            sort_name=db_row["sort_name"],
1982            favorite=bool(db_row["favorite"]),
1983            provider_mappings=provider_mappings,
1984            available=self._summary_available(provider_mappings),
1985            metadata=self._parse_summary_metadata(db_row),
1986        )
1987
1988    @final
1989    @staticmethod
1990    def _parse_summary_provider_mappings(db_row: Mapping[str, Any]) -> set[ProviderMapping]:
1991        """Hydrate the provider mappings of a summary row into ProviderMapping objects."""
1992        if not (raw_mappings := db_row["provider_mappings"]):
1993            return set()
1994        return {ProviderMapping.from_dict(x) for x in json_loads(raw_mappings)}
1995
1996    @final
1997    @staticmethod
1998    def _summary_available(provider_mappings: set[ProviderMapping]) -> bool:
1999        """Compute the availability flag from a summary item's provider mappings."""
2000        # same semantics as the MediaItem.available property
2001        if not (available_providers := get_global_cache_value("available_providers")):
2002            return any(x.available for x in provider_mappings)
2003        if TYPE_CHECKING:
2004            available_providers = cast("set[str]", available_providers)
2005        return any(
2006            x.available and x.provider_instance in available_providers for x in provider_mappings
2007        )
2008
2009    @final
2010    @staticmethod
2011    def _parse_summary_metadata(db_row: Mapping[str, Any]) -> MediaItemMetadataSummary:
2012        """Build the slim metadata of a summary row, carrying only the (first) thumb image."""
2013        thumb: MediaItemImage | None = None
2014        if raw_images := db_row["images"]:
2015            for image in json_loads(raw_images):
2016                if image["type"] != ImageType.THUMB.value:
2017                    continue
2018                thumb = MediaItemImage(
2019                    type=ImageType.THUMB,
2020                    path=image["path"],
2021                    provider=image["provider"],
2022                    remotely_accessible=image.get("remotely_accessible", False),
2023                )
2024                break
2025        return MediaItemMetadataSummary(images=UniqueList([thumb]) if thumb else None)
2026
2027    @final
2028    def _parse_summary_artist_mappings(
2029        self, db_row: Mapping[str, Any]
2030    ) -> UniqueList[ItemMappingSummary]:
2031        """Parse the aggregated slim artist mapping rows of a summary db row."""
2032        return UniqueList(
2033            ItemMappingSummary(
2034                media_type=MediaType.ARTIST,
2035                item_id=str(raw_mapping["item_id"]),
2036                provider="library",
2037                name=raw_mapping["name"],
2038                sort_name=raw_mapping["sort_name"],
2039            )
2040            for raw_mapping in json_loads(db_row["artists"])
2041        )
2042
2043    async def _adapt_query_for_collections(
2044        self,
2045        sql_query: str,
2046        query_params: dict[str, Any],
2047        summary: bool,
2048        order_by: str | None,
2049        collection_name: str | None = None,
2050        search: str | None = None,
2051    ) -> str:
2052        cache_key_json_object = f"collection_{self.api_base}"
2053        json_object = await self.mass.cache.get(key=cache_key_json_object, category=int(summary))
2054        if json_object is None:
2055            # get column names of base query
2056            db_rows = await self.mass.music.database.get_rows_from_query(
2057                sql_query, query_params, limit=1, offset=0
2058            )
2059            # create a sql json_object which queries all these columns
2060            if db_rows:
2061                json_object = (
2062                    "json_object(" + ",".join([f"'{x}',{x}" for x in db_rows[0].keys()]) + ")"  # noqa: SIM118
2063                )
2064                await self.mass.cache.set(
2065                    key=cache_key_json_object, category=int(summary), data=json_object
2066                )
2067            else:
2068                json_object = "json_object()"
2069
2070        collections_column = "collections" if summary else "json_extract(metadata, '$.collections')"
2071
2072        supported_order_keys = [
2073            "name",
2074            "name_desc",
2075            "sort_name",
2076            "sort_name_desc",
2077            "timestamp_added",
2078            "timestamp_added_desc",
2079            "timestamp_modified",
2080            "timestamp_modified_desc",
2081            "last_played",
2082            "last_played_desc",
2083            "play_count",
2084            "play_count_desc",
2085        ]
2086
2087        # additional order options subject to media type
2088        # single is targeting a single media item, collection the aggregated ones
2089        single_extra_order_keys = ""
2090        collection_extra_order_keys = ""
2091        if MediaType.AUDIOBOOK.value in self.api_base:
2092            single_extra_order_keys = "duration,"
2093            collection_extra_order_keys = "SUM(duration) as duration,"
2094            supported_order_keys += ["duration", "duration_desc"]
2095
2096        sql_query = f"""
2097        SELECT * FROM (
2098
2099            WITH
2100                joined_table as ({sql_query}),
2101                collection_extract as (
2102                    SELECT
2103                        name as media_name,
2104                        timestamp_added,
2105                        timestamp_modified,
2106                        last_played,
2107                        play_count,
2108                        {single_extra_order_keys}
2109                        json_extract(iter_coll.value, '$.title') as collection_title,
2110                        json_extract(iter_coll.value, '$.sequence') as collection_sequence,
2111                        json_extract(iter_coll.value, '$.search_title') as collection_search_title,
2112                        json_extract(iter_coll.value, '$.search_sort_title') as collection_search_sort_title,
2113                        CASE
2114                            WHEN json_type(iter_coll.value, '$.sequence') IN ('integer', 'real')
2115                            THEN 1
2116                            WHEN json_type(iter_coll.value, '$.sequence') = 'text'
2117                                AND json_valid(json_extract(iter_coll.value, '$.sequence'))
2118                            THEN CASE
2119                                WHEN json_type(json_extract(iter_coll.value, '$.sequence'))
2120                                    IN ('integer', 'real')
2121                                THEN 1
2122                                ELSE 0
2123                            END
2124                            ELSE 0
2125                        END as collection_sequence_is_numeric,
2126                        {json_object} as media_data
2127                    FROM (
2128                        SELECT * FROM joined_table
2129                    ), json_each({collections_column}) as iter_coll
2130                )
2131            SELECT
2132                'collection' as type,
2133                collection_title as name,
2134                COALESCE(MAX(collection_search_title), replace(lower(collection_title),' ','')) AS search_name,
2135                COALESCE(MAX(collection_search_sort_title), replace(lower(collection_title),' ','')) AS search_sort_name,
2136                MAX(timestamp_added) as timestamp_added,
2137                MAX(timestamp_modified) as timestamp_modified,
2138                MAX(last_played) as last_played,
2139                SUM(play_count) as play_count,
2140                {collection_extra_order_keys}
2141                json_group_array(media_data) as media_data
2142            FROM (
2143                SELECT * FROM collection_extract
2144                -- NOTE: The following ORDER_BY to control the aggregation order of json_group_array is undocumented sqlite behavior
2145                -- Confirmed working with sqlite 3.40.1 & 3.53
2146                -- Once our image moves to sqlite 3.44 we can and should make use of ORDER_BY in the aggregate itself
2147                ORDER BY collection_title,
2148                -- null case
2149                CASE WHEN collection_sequence IS NULL THEN 1 ELSE 0 END,
2150                -- numeric before text
2151                CASE WHEN collection_sequence_is_numeric THEN 0 ELSE 1 END,
2152                -- order NUMERIC
2153                CASE WHEN collection_sequence_is_numeric
2154                    THEN CAST(collection_sequence AS REAL)
2155                END,
2156                -- order TEXT
2157                CASE WHEN NOT collection_sequence_is_numeric
2158                    THEN collection_sequence
2159                END COLLATE NOCASE,
2160                -- order by media name if no sequence given
2161                CASE
2162                    WHEN collection_sequence IS NULL
2163                    THEN media_name
2164                END COLLATE NOCASE
2165            )
2166            GROUP BY collection_title
2167
2168            UNION ALL
2169
2170            SELECT 'single', name, search_name, search_sort_name,
2171                timestamp_added, timestamp_modified, last_played, play_count,
2172                {single_extra_order_keys}
2173                {json_object} FROM joined_table
2174                WHERE {collections_column} IS NULL
2175                    OR {collections_column} = '[]'
2176        )
2177        """
2178
2179        if collection_name:
2180            sql_query += " WHERE type = 'collection' AND name = :collection_name"
2181            return sql_query
2182
2183        if search:
2184            sql_query += " WHERE search_name LIKE :search"
2185
2186        if order_by:
2187            if order_by not in supported_order_keys:
2188                self.logger.warning("%s is not supported for order_by key in collections", order_by)
2189                order_by = "name"  # fallback
2190            if sort_key := SORT_KEYS.get(order_by):
2191                sql_query += f" ORDER BY {sort_key}"
2192
2193        return sql_query
2194