From: Marcel van der Veldt Date: Sun, 4 Jan 2026 01:00:30 +0000 (+0100) Subject: Fix sql injection vulnerability (#2916) X-Git-Url: https://git.kitaultman.com/?a=commitdiff_plain;h=088b2a9b6a14de84e5a61e456bfa1bae05749e2e;p=music-assistant-server.git Fix sql injection vulnerability (#2916) --- diff --git a/music_assistant/controllers/media/albums.py b/music_assistant/controllers/media/albums.py index 5e8b4252..c399c487 100644 --- a/music_assistant/controllers/media/albums.py +++ b/music_assistant/controllers/media/albums.py @@ -112,8 +112,6 @@ class AlbumsController(MediaControllerBase[Album]): offset: int = 0, order_by: str = "sort_name", provider: str | list[str] | None = None, - extra_query: str | None = None, - extra_query_params: dict[str, Any] | None = None, album_types: list[AlbumType] | None = None, ) -> list[Album]: """Get in-database albums. @@ -124,12 +122,10 @@ class AlbumsController(MediaControllerBase[Album]): :param offset: Number of items to skip. :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added'). :param provider: Filter by provider instance ID (single string or list). - :param extra_query: Additional SQL query string. - :param extra_query_params: Additional query parameters. :param album_types: Filter by album types. """ - extra_query_params = extra_query_params or {} - extra_query_parts: list[str] = [extra_query] if extra_query else [] + extra_query_params: dict[str, Any] = {} + extra_query_parts: list[str] = [] extra_join_parts: list[str] = [] artist_table_joined = False # optional album type filter @@ -161,7 +157,7 @@ class AlbumsController(MediaControllerBase[Album]): ) artist_table_joined = True extra_query_params["search_artist"] = f"%{artist_str}%" - result = await self._get_library_items_by_query( + result = await self.get_library_items_by_query( favorite=favorite, search=search, limit=limit, @@ -189,7 +185,7 @@ class AlbumsController(MediaControllerBase[Album]): extra_query_params["search_artist"] = f"%{search}%" existing_uris = {item.uri for item in result} - for album in await self._get_library_items_by_query( + for album in await self.get_library_items_by_query( favorite=favorite, search=None, limit=remaining_limit, @@ -360,8 +356,10 @@ class AlbumsController(MediaControllerBase[Album]): item_id: str | int, ) -> list[Track]: """Return in-database album tracks for the given database album.""" - return await self.mass.music.tracks._get_library_items_by_query( - extra_query_parts=[f"WHERE album_tracks.album_id = {item_id}"], + db_id = int(item_id) # ensure integer + return await self.mass.music.tracks.get_library_items_by_query( + extra_query_parts=["WHERE album_tracks.album_id = :album_id"], + extra_query_params={"album_id": db_id}, ) async def add_item_mapping_as_album_to_library(self, item: ItemMapping) -> Album: diff --git a/music_assistant/controllers/media/artists.py b/music_assistant/controllers/media/artists.py index a1a65153..91566270 100644 --- a/music_assistant/controllers/media/artists.py +++ b/music_assistant/controllers/media/artists.py @@ -71,8 +71,6 @@ class ArtistsController(MediaControllerBase[Artist]): offset: int = 0, order_by: str = "sort_name", provider: str | list[str] | None = None, - extra_query: str | None = None, - extra_query_params: dict[str, Any] | None = None, album_artists_only: bool = False, ) -> list[Artist]: """Get in-database (album) artists. @@ -83,18 +81,16 @@ class ArtistsController(MediaControllerBase[Artist]): :param offset: Number of items to skip. :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added'). :param provider: Filter by provider instance ID (single string or list). - :param extra_query: Additional SQL query string. - :param extra_query_params: Additional query parameters. :param album_artists_only: Only return artists that have albums. """ - extra_query_params = extra_query_params or {} - extra_query_parts: list[str] = [extra_query] if extra_query else [] + extra_query_params: dict[str, Any] = {} + extra_query_parts: list[str] = [] if album_artists_only: extra_query_parts.append( f"artists.item_id in (select {DB_TABLE_ALBUM_ARTISTS}.artist_id " f"from {DB_TABLE_ALBUM_ARTISTS})" ) - return await self._get_library_items_by_query( + return await self.get_library_items_by_query( favorite=favorite, search=search, limit=limit, @@ -202,7 +198,8 @@ class ArtistsController(MediaControllerBase[Artist]): # recursively also remove artist albums for db_row in await self.mass.music.database.get_rows_from_query( - f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = {db_id}", + f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = :artist_id", + {"artist_id": db_id}, limit=5000, ): if not recursive: @@ -211,7 +208,8 @@ class ArtistsController(MediaControllerBase[Artist]): await self.mass.music.albums.remove_item_from_library(db_row["album_id"]) # recursively also remove artist tracks for db_row in await self.mass.music.database.get_rows_from_query( - f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = {db_id}", + f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = :artist_id", + {"artist_id": db_id}, limit=5000, ): if not recursive: @@ -240,13 +238,13 @@ class ArtistsController(MediaControllerBase[Artist]): item_id, provider_instance_id_or_domain, ): - artist_id = db_artist.item_id - subquery = ( - f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = {artist_id}" - ) + db_artist_id = int(db_artist.item_id) # ensure integer + subquery = f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = :artist_id" query = f"tracks.item_id in ({subquery})" - return await self.mass.music.tracks._get_library_items_by_query( - extra_query_parts=[query], provider_filter=[provider_instance_id_or_domain] + return await self.mass.music.tracks.get_library_items_by_query( + extra_query_parts=[query], + extra_query_params={"artist_id": db_artist_id}, + provider_filter=[provider_instance_id_or_domain], ) return [] @@ -255,9 +253,13 @@ class ArtistsController(MediaControllerBase[Artist]): item_id: str | int, ) -> list[Track]: """Return all tracks for an artist in the library/db.""" - subquery = f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = {item_id}" + db_id = int(item_id) # ensure integer + subquery = f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = :artist_id" query = f"tracks.item_id in ({subquery})" - return await self.mass.music.tracks.library_items(extra_query=query) + return await self.mass.music.tracks.get_library_items_by_query( + extra_query_parts=[query], + extra_query_params={"artist_id": db_id}, + ) async def get_provider_artist_albums( self, @@ -276,13 +278,13 @@ class ArtistsController(MediaControllerBase[Artist]): item_id, provider_instance_id_or_domain, ): - artist_id = db_artist.item_id - subquery = ( - f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = {artist_id}" - ) + db_artist_id = int(db_artist.item_id) # ensure integer + subquery = f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = :artist_id" query = f"albums.item_id in ({subquery})" - return await self.mass.music.albums._get_library_items_by_query( - extra_query_parts=[query], provider_filter=[provider_instance_id_or_domain] + return await self.mass.music.albums.get_library_items_by_query( + extra_query_parts=[query], + extra_query_params={"artist_id": db_artist_id}, + provider_filter=[provider_instance_id_or_domain], ) return [] @@ -291,9 +293,13 @@ class ArtistsController(MediaControllerBase[Artist]): item_id: str | int, ) -> list[Album]: """Return all in-library albums for an artist.""" - subquery = f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = {item_id}" + db_id = int(item_id) # ensure integer + subquery = f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = :artist_id" query = f"albums.item_id in ({subquery})" - return await self.mass.music.albums.library_items(extra_query=query) + return await self.mass.music.albums.get_library_items_by_query( + extra_query_parts=[query], + extra_query_params={"artist_id": db_id}, + ) async def _add_library_item( self, item: Artist | ItemMapping, overwrite_existing: bool = False diff --git a/music_assistant/controllers/media/audiobooks.py b/music_assistant/controllers/media/audiobooks.py index a865cd97..d908d3a7 100644 --- a/music_assistant/controllers/media/audiobooks.py +++ b/music_assistant/controllers/media/audiobooks.py @@ -68,8 +68,6 @@ class AudiobooksController(MediaControllerBase[Audiobook]): offset: int = 0, order_by: str = "sort_name", provider: str | list[str] | None = None, - extra_query: str | None = None, - extra_query_params: dict[str, Any] | None = None, ) -> list[Audiobook]: """Get in-database audiobooks. @@ -79,12 +77,10 @@ class AudiobooksController(MediaControllerBase[Audiobook]): :param offset: Number of items to skip. :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added'). :param provider: Filter by provider instance ID (single string or list). - :param extra_query: Additional SQL query string. - :param extra_query_params: Additional query parameters. """ - extra_query_params = extra_query_params or {} - extra_query_parts: list[str] = [extra_query] if extra_query else [] - result = await self._get_library_items_by_query( + extra_query_params: dict[str, Any] = {} + extra_query_parts: list[str] = [] + result = await self.get_library_items_by_query( favorite=favorite, search=search, limit=limit, @@ -100,7 +96,7 @@ class AudiobooksController(MediaControllerBase[Audiobook]): "WHERE audiobooks.authors LIKE :search or audiobooks.narrators LIKE :search", ] extra_query_params["search"] = f"%{search}%" - return result + await self._get_library_items_by_query( + return result + await self.get_library_items_by_query( favorite=favorite, search=None, limit=limit, diff --git a/music_assistant/controllers/media/base.py b/music_assistant/controllers/media/base.py index fb1e9fa8..01d0915d 100644 --- a/music_assistant/controllers/media/base.py +++ b/music_assistant/controllers/media/base.py @@ -166,7 +166,7 @@ class MediaControllerBase[ItemCls: "MediaItemType"](metaclass=ABCMeta): # search by (exact) name match query = f"{self.db_table}.name = :name OR {self.db_table}.sort_name = :sort_name" query_params = {"name": item.name, "sort_name": item.sort_name} - for db_item in await self._get_library_items_by_query( + for db_item in await self.get_library_items_by_query( extra_query_parts=[query], extra_query_params=query_params ): if compare_media_item(db_item, item, True): @@ -242,19 +242,24 @@ class MediaControllerBase[ItemCls: "MediaItemType"](metaclass=ABCMeta): offset: int = 0, order_by: str = "sort_name", provider: str | list[str] | None = None, - extra_query: str | None = None, - extra_query_params: dict[str, Any] | None = None, ) -> list[ItemCls]: - """Get in-database items.""" - return await self._get_library_items_by_query( + """ + Get the library items for this mediatype. + + :param favorite: Filter by favorite status. + :param search: Filter by search query. + :param limit: Maximum number of items to return. + :param offset: Number of items to skip. + :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added'). + :param provider: Filter by provider instance ID (single string or list). + """ + return await self.get_library_items_by_query( favorite=favorite, search=search, limit=limit, offset=offset, order_by=order_by, provider_filter=self._ensure_provider_filter(provider), - extra_query_parts=[extra_query] if extra_query else None, - extra_query_params=extra_query_params, ) async def iter_library_items( @@ -263,8 +268,6 @@ class MediaControllerBase[ItemCls: "MediaItemType"](metaclass=ABCMeta): search: str | None = None, order_by: str = "sort_name", provider: str | list[str] | None = None, - extra_query: str | None = None, - extra_query_params: dict[str, Any] | None = None, ) -> AsyncGenerator[ItemCls, None]: """Iterate all in-database items.""" limit: int = 500 @@ -274,15 +277,13 @@ class MediaControllerBase[ItemCls: "MediaItemType"](metaclass=ABCMeta): else: provider_filter = None while True: - next_items = await self._get_library_items_by_query( + next_items = await self.get_library_items_by_query( favorite=favorite, search=search, limit=limit, offset=offset, order_by=order_by, provider_filter=provider_filter, - extra_query_parts=[extra_query] if extra_query else None, - extra_query_params=extra_query_params, ) for item in next_items: yield item @@ -357,9 +358,10 @@ class MediaControllerBase[ItemCls: "MediaItemType"](metaclass=ABCMeta): async def get_library_item(self, item_id: int | str) -> ItemCls: """Get single library item by id.""" db_id = int(item_id) # ensure integer - extra_query = f"WHERE {self.db_table}.item_id = {item_id}" - for db_item in await self._get_library_items_by_query( + extra_query = f"WHERE {self.db_table}.item_id = :item_id" + for db_item in await self.get_library_items_by_query( extra_query_parts=[extra_query], + extra_query_params={"item_id": db_id}, ): return db_item msg = f"{self.media_type.value} not found in library: {db_id}" @@ -414,7 +416,7 @@ class MediaControllerBase[ItemCls: "MediaItemType"](metaclass=ABCMeta): external_id_str = f'%"{external_id_type}","{external_id}"%' else: external_id_str = f'%"{external_id}"%' - for item in await self._get_library_items_by_query( + for item in await self.get_library_items_by_query( extra_query_parts=[query], extra_query_params={"external_id_str": external_id_str}, ): @@ -464,7 +466,7 @@ class MediaControllerBase[ItemCls: "MediaItemType"](metaclass=ABCMeta): query_params["item_id"] = provider_item_id subquery = f"SELECT item_id FROM provider_mappings WHERE {' AND '.join(subquery_parts)}" query = f"WHERE {self.db_table}.item_id IN ({subquery})" - return await self._get_library_items_by_query( + return await self.get_library_items_by_query( limit=limit, offset=offset, extra_query_parts=[query], @@ -767,7 +769,7 @@ class MediaControllerBase[ItemCls: "MediaItemType"](metaclass=ABCMeta): """ @final - async def _get_library_items_by_query( + async def get_library_items_by_query( self, favorite: bool | None = None, search: str | None = None, @@ -892,12 +894,15 @@ class MediaControllerBase[ItemCls: "MediaItemType"](metaclass=ABCMeta): # Apply the provider filter if provider_filter: provider_conditions = [] - for prov in provider_filter: - provider_conditions.append(f"provider_mappings.provider_instance = '{prov}'") + for idx, prov in enumerate(provider_filter): + param_name = f"provider_filter_{idx}" + provider_conditions.append(f"provider_mappings.provider_instance = :{param_name}") + query_params[param_name] = prov + query_params["provider_media_type"] = self.media_type.value join_parts.append( f"JOIN provider_mappings ON provider_mappings.item_id = {self.db_table}.item_id " - f"AND provider_mappings.media_type = '{self.media_type.value}' " - f"AND provider_mappings.in_library = 1 " + "AND provider_mappings.media_type = :provider_media_type " + "AND provider_mappings.in_library = 1 " f"AND ({' OR '.join(provider_conditions)})" ) diff --git a/music_assistant/controllers/media/podcasts.py b/music_assistant/controllers/media/podcasts.py index 7530b958..8361ec23 100644 --- a/music_assistant/controllers/media/podcasts.py +++ b/music_assistant/controllers/media/podcasts.py @@ -50,8 +50,6 @@ class PodcastsController(MediaControllerBase[Podcast]): offset: int = 0, order_by: str = "sort_name", provider: str | list[str] | None = None, - extra_query: str | None = None, - extra_query_params: dict[str, Any] | None = None, ) -> list[Podcast]: """Get in-database podcasts. @@ -61,28 +59,24 @@ class PodcastsController(MediaControllerBase[Podcast]): :param offset: Number of items to skip. :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added'). :param provider: Filter by provider instance ID (single string or list). - :param extra_query: Additional SQL query string. - :param extra_query_params: Additional query parameters. """ - extra_query_params = extra_query_params or {} - extra_query_parts: list[str] = [extra_query] if extra_query else [] - result = await self._get_library_items_by_query( + result = await self.get_library_items_by_query( favorite=favorite, search=search, limit=limit, offset=offset, order_by=order_by, provider_filter=self._ensure_provider_filter(provider), - extra_query_parts=extra_query_parts, - extra_query_params=extra_query_params, ) if search and len(result) < 25 and not offset: # append publisher items to result - extra_query_parts = [ + extra_query_parts: list[str] = [ "WHERE podcasts.publisher LIKE :search", ] - extra_query_params["search"] = f"%{search}%" - return result + await self._get_library_items_by_query( + extra_query_params: dict[str, Any] = { + "search": f"%{search}%", + } + return result + await self.get_library_items_by_query( favorite=favorite, search=None, limit=limit, diff --git a/music_assistant/controllers/media/tracks.py b/music_assistant/controllers/media/tracks.py index 91d9e038..542ff2b4 100644 --- a/music_assistant/controllers/media/tracks.py +++ b/music_assistant/controllers/media/tracks.py @@ -166,8 +166,6 @@ class TracksController(MediaControllerBase[Track]): offset: int = 0, order_by: str = "sort_name", provider: str | list[str] | None = None, - extra_query: str | None = None, - extra_query_params: dict[str, Any] | None = None, ) -> list[Track]: """Get in-database tracks. @@ -177,11 +175,9 @@ class TracksController(MediaControllerBase[Track]): :param offset: Number of items to skip. :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added'). :param provider: Filter by provider instance ID (single string or list). - :param extra_query: Additional SQL query string. - :param extra_query_params: Additional query parameters. """ - extra_query_params = extra_query_params or {} - extra_query_parts: list[str] = [extra_query] if extra_query else [] + extra_query_params: dict[str, Any] = {} + extra_query_parts: list[str] = [] extra_join_parts: list[str] = [] if search and " - " in search: # handle combined artist + title search @@ -198,7 +194,7 @@ class TracksController(MediaControllerBase[Track]): "AND artists.search_name LIKE :search_artist" ) extra_query_params["search_artist"] = f"%{artist_str}%" - result = await self._get_library_items_by_query( + result = await self.get_library_items_by_query( favorite=favorite, search=search, limit=limit, @@ -219,7 +215,7 @@ class TracksController(MediaControllerBase[Track]): ) extra_query_params["search_artist"] = f"%{artist_search_str}%" existing_uris = {item.uri for item in result} - for _track in await self._get_library_items_by_query( + for _track in await self.get_library_items_by_query( favorite=favorite, search=None, limit=limit, @@ -410,12 +406,16 @@ class TracksController(MediaControllerBase[Track]): item_id: str | int, ) -> list[Album]: """Return all in-library albums for a track.""" + db_id = int(item_id) # ensure integer subquery = ( f"SELECT album_id FROM {DB_TABLE_ALBUM_TRACKS} " - f"WHERE {DB_TABLE_ALBUM_TRACKS}.track_id = {item_id}" + f"WHERE {DB_TABLE_ALBUM_TRACKS}.track_id = :track_id" ) query = f"{DB_TABLE_ALBUMS}.item_id in ({subquery})" - return await self.mass.music.albums._get_library_items_by_query(extra_query_parts=[query]) + return await self.mass.music.albums.get_library_items_by_query( + extra_query_parts=[query], + extra_query_params={"track_id": db_id}, + ) async def match_provider( self, diff --git a/music_assistant/controllers/metadata.py b/music_assistant/controllers/metadata.py index c071c0a3..89c863ba 100644 --- a/music_assistant/controllers/metadata.py +++ b/music_assistant/controllers/metadata.py @@ -976,8 +976,8 @@ class MetaDataController(CoreController): f"AND (json_extract({DB_TABLE_ARTISTS}.metadata,'$.images') ISNULL " f"OR json_extract({DB_TABLE_ARTISTS}.metadata,'$.images') = '[]')" ) - for artist in await self.mass.music.artists.library_items( - limit=5, order_by="random", extra_query=query + for artist in await self.mass.music.artists.get_library_items_by_query( + limit=5, order_by="random", extra_query_parts=[query] ): if artist.uri: self.schedule_update_metadata(artist.uri) @@ -990,8 +990,8 @@ class MetaDataController(CoreController): f"json_extract({DB_TABLE_PLAYLISTS}.metadata,'$.last_refresh') ISNULL " f"OR json_extract({DB_TABLE_PLAYLISTS}.metadata,'$.last_refresh') < {timestamp}" ) - for playlist in await self.mass.music.playlists.library_items( - limit=5, order_by="random", extra_query=query + for playlist in await self.mass.music.playlists.get_library_items_by_query( + limit=5, order_by="random", extra_query_parts=[query] ): if playlist.uri: self.schedule_update_metadata(playlist.uri) diff --git a/music_assistant/providers/builtin/__init__.py b/music_assistant/providers/builtin/__init__.py index 82f3324c..e0490e2d 100644 --- a/music_assistant/providers/builtin/__init__.py +++ b/music_assistant/providers/builtin/__init__.py @@ -554,8 +554,11 @@ class BuiltinProvider(MusicProvider): @use_cache(expiration=3600, category=CACHE_CATEGORY_PLAYLISTS) async def _get_builtin_playlist_random_album(self) -> list[Track]: - for random_album in await self.mass.music.albums.library_items( - limit=1, order_by="random", extra_query="album_type != 'single'" + for random_album in await self.mass.music.albums.get_library_items_by_query( + limit=1, + order_by="random", + extra_query_parts=["album_type != :excluded_album_type"], + extra_query_params={"excluded_album_type": "single"}, ): tracks = await self.mass.music.albums.tracks( random_album.item_id, random_album.provider diff --git a/music_assistant/providers/filesystem_local/__init__.py b/music_assistant/providers/filesystem_local/__init__.py index 3538a83f..e7c81a89 100644 --- a/music_assistant/providers/filesystem_local/__init__.py +++ b/music_assistant/providers/filesystem_local/__init__.py @@ -232,37 +232,37 @@ class LocalFileSystemProvider(MusicProvider): # searching the filesystem is slow and unreliable, # so instead we just query the db... if media_types is None or MediaType.TRACK in media_types: - result.tracks = await self.mass.music.tracks._get_library_items_by_query( + result.tracks = await self.mass.music.tracks.get_library_items_by_query( search=search_query, provider_filter=[self.instance_id], limit=limit ) if media_types is None or MediaType.ALBUM in media_types: - result.albums = await self.mass.music.albums._get_library_items_by_query( + result.albums = await self.mass.music.albums.get_library_items_by_query( search=search_query, provider_filter=[self.instance_id], limit=limit, ) if media_types is None or MediaType.ARTIST in media_types: - result.artists = await self.mass.music.artists._get_library_items_by_query( + result.artists = await self.mass.music.artists.get_library_items_by_query( search=search_query, provider_filter=[self.instance_id], limit=limit, ) if media_types is None or MediaType.PLAYLIST in media_types: - result.playlists = await self.mass.music.playlists._get_library_items_by_query( + result.playlists = await self.mass.music.playlists.get_library_items_by_query( search=search_query, provider_filter=[self.instance_id], limit=limit, ) if media_types is None or MediaType.AUDIOBOOK in media_types: - result.audiobooks = await self.mass.music.audiobooks._get_library_items_by_query( + result.audiobooks = await self.mass.music.audiobooks.get_library_items_by_query( search=search_query, provider_filter=[self.instance_id], limit=limit, ) if media_types is None or MediaType.PODCAST in media_types: - result.podcasts = await self.mass.music.podcasts._get_library_items_by_query( + result.podcasts = await self.mass.music.podcasts.get_library_items_by_query( search=search_query, provider_filter=[self.instance_id], limit=limit, diff --git a/music_assistant/providers/plex/__init__.py b/music_assistant/providers/plex/__init__.py index 778bdefb..2bb5c57b 100644 --- a/music_assistant/providers/plex/__init__.py +++ b/music_assistant/providers/plex/__init__.py @@ -567,7 +567,7 @@ class PlexProvider(MusicProvider): ) async def _get_or_create_artist_by_name(self, artist_name: str) -> Artist | ItemMapping: - if library_items := await self.mass.music.artists._get_library_items_by_query( + if library_items := await self.mass.music.artists.get_library_items_by_query( search=artist_name, provider_filter=[self.instance_id] ): return ItemMapping.from_item(library_items[0])