/
/
/
1"""Manage MediaItems of type Track."""
2
3from __future__ import annotations
4
5import urllib.parse
6from collections.abc import Iterable
7from typing import TYPE_CHECKING, Any
8
9from music_assistant_models.enums import MediaType, ProviderFeature
10from music_assistant_models.errors import (
11 InvalidDataError,
12 MusicAssistantError,
13 UnsupportedFeaturedException,
14)
15from music_assistant_models.media_items import (
16 Album,
17 Artist,
18 ItemMapping,
19 ProviderMapping,
20 Track,
21 UniqueList,
22)
23
24from music_assistant.constants import (
25 DB_TABLE_ALBUM_TRACKS,
26 DB_TABLE_ALBUMS,
27 DB_TABLE_TRACK_ARTISTS,
28 DB_TABLE_TRACKS,
29)
30from music_assistant.helpers.compare import (
31 compare_artists,
32 compare_media_item,
33 compare_track,
34 create_safe_string,
35 loose_compare_strings,
36)
37from music_assistant.helpers.database import UNSET
38from music_assistant.helpers.json import serialize_to_json
39from music_assistant.models.music_provider import MusicProvider
40
41from .base import MediaControllerBase
42
43if TYPE_CHECKING:
44 from music_assistant import MusicAssistant
45
46
47class TracksController(MediaControllerBase[Track]):
48 """Controller managing MediaItems of type Track."""
49
50 db_table = DB_TABLE_TRACKS
51 media_type = MediaType.TRACK
52 item_cls = Track
53
54 def __init__(self, mass: MusicAssistant) -> None:
55 """Initialize class."""
56 super().__init__(mass)
57 self.base_query = """
58 SELECT
59 tracks.*,
60 (SELECT JSON_GROUP_ARRAY(
61 json_object(
62 'item_id', track_pm.provider_item_id,
63 'provider_domain', track_pm.provider_domain,
64 'provider_instance', track_pm.provider_instance,
65 'available', track_pm.available,
66 'audio_format', json(track_pm.audio_format),
67 'url', track_pm.url,
68 'details', track_pm.details,
69 'in_library', track_pm.in_library,
70 'is_unique', track_pm.is_unique
71 )) FROM provider_mappings track_pm WHERE track_pm.item_id = tracks.item_id AND track_pm.media_type = 'track') AS provider_mappings,
72
73 (SELECT JSON_GROUP_ARRAY(
74 json_object(
75 'item_id', artists.item_id,
76 'provider', 'library',
77 'name', artists.name,
78 'sort_name', artists.sort_name,
79 'media_type', 'artist'
80 )) FROM artists JOIN track_artists on track_artists.track_id = tracks.item_id WHERE artists.item_id = track_artists.artist_id) AS artists,
81 (SELECT
82 json_object(
83 'item_id', albums.item_id,
84 'provider', 'library',
85 'name', albums.name,
86 'sort_name', albums.sort_name,
87 'media_type', 'album',
88 'disc_number', album_tracks.disc_number,
89 'track_number', album_tracks.track_number,
90 'images', json_extract(albums.metadata, '$.images')
91 ) FROM albums WHERE albums.item_id = album_tracks.album_id) AS track_album
92 FROM tracks
93 LEFT JOIN album_tracks on album_tracks.track_id = tracks.item_id
94 """ # noqa: E501
95 # register (extra) api handlers
96 api_base = self.api_base
97 self.mass.register_api_command(f"music/{api_base}/track_versions", self.versions)
98 self.mass.register_api_command(f"music/{api_base}/track_albums", self.albums)
99 self.mass.register_api_command(f"music/{api_base}/preview", self.get_preview_url)
100 self.mass.register_api_command(f"music/{api_base}/similar_tracks", self.similar_tracks)
101
102 async def get(
103 self,
104 item_id: str,
105 provider_instance_id_or_domain: str,
106 recursive: bool = True,
107 album_uri: str | None = None,
108 ) -> Track:
109 """Return (full) details for a single media item."""
110 track = await super().get(
111 item_id,
112 provider_instance_id_or_domain,
113 )
114 if not recursive and album_uri is None:
115 # return early if we do not want recursive full details and no album uri is provided
116 return track
117
118 # append full album details to full track item (resolve ItemMappings)
119 try:
120 if album_uri:
121 item = await self.mass.music.get_item_by_uri(album_uri)
122 if isinstance(item, Album):
123 track.album = item
124 elif provider_instance_id_or_domain == "library":
125 # grab the first album this track is attached to
126 for album_track_row in await self.mass.music.database.get_rows(
127 DB_TABLE_ALBUM_TRACKS, {"track_id": int(item_id)}, limit=1
128 ):
129 track.album = await self.mass.music.albums.get_library_item(
130 album_track_row["album_id"]
131 )
132 elif isinstance(track.album, ItemMapping) or (track.album and not track.album.image):
133 track.album = await self.mass.music.albums.get(
134 track.album.item_id, track.album.provider, recursive=False
135 )
136 except MusicAssistantError as err:
137 # edge case where playlist track has invalid albumdetails
138 self.logger.warning("Unable to fetch album details for %s - %s", track.uri, str(err))
139
140 if not recursive:
141 return track
142
143 # append artist details to full track item (resolve ItemMappings)
144 track_artists = []
145 for artist in track.artists:
146 if not isinstance(artist, ItemMapping):
147 track_artists.append(artist)
148 continue
149 try:
150 track_artists.append(
151 await self.mass.music.artists.get(
152 artist.item_id,
153 artist.provider,
154 )
155 )
156 except MusicAssistantError as err:
157 # edge case where playlist track has invalid artistdetails
158 self.logger.warning("Unable to fetch artist details %s - %s", artist.uri, str(err))
159 track.artists = UniqueList(track_artists)
160 return track
161
162 async def library_items(
163 self,
164 favorite: bool | None = None,
165 search: str | None = None,
166 limit: int = 500,
167 offset: int = 0,
168 order_by: str = "sort_name",
169 provider: str | list[str] | None = None,
170 ) -> list[Track]:
171 """Get in-database tracks.
172
173 :param favorite: Filter by favorite status.
174 :param search: Filter by search query.
175 :param limit: Maximum number of items to return.
176 :param offset: Number of items to skip.
177 :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
178 :param provider: Filter by provider instance ID (single string or list).
179 """
180 extra_query_params: dict[str, Any] = {}
181 extra_query_parts: list[str] = []
182 extra_join_parts: list[str] = []
183 if search and " - " in search:
184 # handle combined artist + title search
185 artist_str, title_str = search.split(" - ", 1)
186 search = None
187 title_str = create_safe_string(title_str, True, True)
188 artist_str = create_safe_string(artist_str, True, True)
189 extra_query_parts.append("tracks.search_name LIKE :search_title")
190 extra_query_params["search_title"] = f"%{title_str}%"
191 # use join with artists table to filter on artist name
192 extra_join_parts.append(
193 "JOIN track_artists ON track_artists.track_id = tracks.item_id "
194 "JOIN artists ON artists.item_id = track_artists.artist_id "
195 "AND artists.search_name LIKE :search_artist"
196 )
197 extra_query_params["search_artist"] = f"%{artist_str}%"
198 result = await self.get_library_items_by_query(
199 favorite=favorite,
200 search=search,
201 limit=limit,
202 offset=offset,
203 order_by=order_by,
204 provider_filter=self._ensure_provider_filter(provider),
205 extra_query_parts=extra_query_parts,
206 extra_query_params=extra_query_params,
207 extra_join_parts=extra_join_parts,
208 )
209 if search and len(result) < 25 and not offset:
210 # append artist items to result
211 artist_search_str = create_safe_string(search, True, True)
212 extra_join_parts.append(
213 "JOIN track_artists ON track_artists.track_id = tracks.item_id "
214 "JOIN artists ON artists.item_id = track_artists.artist_id "
215 "AND artists.search_name LIKE :search_artist"
216 )
217 extra_query_params["search_artist"] = f"%{artist_search_str}%"
218 existing_uris = {item.uri for item in result}
219 for _track in await self.get_library_items_by_query(
220 favorite=favorite,
221 search=None,
222 limit=limit,
223 order_by=order_by,
224 provider_filter=self._ensure_provider_filter(provider),
225 extra_query_parts=extra_query_parts,
226 extra_query_params=extra_query_params,
227 extra_join_parts=extra_join_parts,
228 ):
229 # prevent duplicates (when artist is also in the title)
230 if _track.uri not in existing_uris:
231 result.append(_track)
232 return result
233
234 async def versions(
235 self,
236 item_id: str,
237 provider_instance_id_or_domain: str,
238 ) -> UniqueList[Track]:
239 """Return all versions of a track we can find on all providers."""
240 track = await self.get(item_id, provider_instance_id_or_domain)
241 search_query = f"{track.artist_str} - {track.name}"
242 result: UniqueList[Track] = UniqueList()
243 for provider_id in self.mass.music.get_unique_providers():
244 provider = self.mass.get_provider(provider_id)
245 if not isinstance(provider, MusicProvider):
246 continue
247 if not provider.library_supported(MediaType.TRACK):
248 continue
249 result.extend(
250 prov_item
251 for prov_item in await self.search(search_query, provider_id)
252 if loose_compare_strings(track.name, prov_item.name)
253 and compare_artists(prov_item.artists, track.artists, any_match=True)
254 # make sure that the 'base' version is NOT included
255 and not track.provider_mappings.intersection(prov_item.provider_mappings)
256 )
257 return result
258
259 async def albums(
260 self,
261 item_id: str,
262 provider_instance_id_or_domain: str,
263 in_library_only: bool = False,
264 ) -> UniqueList[Album]:
265 """Return all albums the track appears on."""
266 full_track = await self.get(item_id, provider_instance_id_or_domain)
267 db_items = (
268 await self.get_library_track_albums(full_track.item_id)
269 if full_track.provider == "library"
270 else []
271 )
272 # return all (unique) items from all providers
273 result: UniqueList[Album] = UniqueList(db_items)
274 # use search to get all items on the provider
275 search_query = f"{full_track.artist_str} - {full_track.name}"
276 # TODO: we could use musicbrainz info here to get a list of all releases known
277 unique_ids: set[str] = set()
278 for prov_item in (await self.mass.music.search(search_query, [MediaType.TRACK])).tracks:
279 if not isinstance(prov_item, Track): # for type checking
280 continue
281 if not loose_compare_strings(full_track.name, prov_item.name):
282 continue
283 if not prov_item.album:
284 continue
285 if not compare_artists(full_track.artists, prov_item.artists, any_match=True):
286 continue
287 unique_id = f"{prov_item.album.name}.{prov_item.album.version}"
288 if unique_id in unique_ids:
289 continue
290 unique_ids.add(unique_id)
291 # prefer db item
292 if db_item := await self.mass.music.albums.get_library_item_by_prov_id(
293 prov_item.album.item_id, prov_item.album.provider
294 ):
295 result.append(db_item)
296 elif not in_library_only and isinstance(prov_item.album, Album):
297 result.append(prov_item.album)
298 return result
299
300 async def similar_tracks(
301 self,
302 item_id: str,
303 provider_instance_id_or_domain: str,
304 limit: int = 25,
305 allow_lookup: bool = False,
306 preferred_provider_instances: list[str] | None = None,
307 ) -> list[Track]:
308 """
309 Get a list of similar tracks for the given track.
310
311 :param item_id: The item ID of the track.
312 :param provider_instance_id_or_domain: The provider instance ID or domain.
313 :param limit: Maximum number of similar tracks to return.
314 :param allow_lookup: Allow lookup on other providers if not found.
315 :param preferred_provider_instances: List of preferred provider instance IDs to use.
316 When provided, these providers will be tried first before falling back to others.
317 """
318 ref_item = await self.get(item_id, provider_instance_id_or_domain)
319
320 # Sort provider mappings to prefer user's provider instances
321 def sort_key(mapping: ProviderMapping) -> tuple[int, int]:
322 # Primary sort: preferred providers first (0), then others (1)
323 preferred = (
324 0
325 if preferred_provider_instances
326 and mapping.provider_instance in preferred_provider_instances
327 else 1
328 )
329 # Secondary sort: by quality (higher is better, so negate)
330 quality = -(mapping.quality or 0)
331 return (preferred, quality)
332
333 sorted_mappings = sorted(ref_item.provider_mappings, key=sort_key)
334
335 # Try preferred providers first, then fall back to others
336 for allow_other_provider in (False, True):
337 for prov_mapping in sorted_mappings:
338 if (
339 not allow_other_provider
340 and preferred_provider_instances
341 and prov_mapping.provider_instance not in preferred_provider_instances
342 ):
343 continue
344 prov = self.mass.get_provider(prov_mapping.provider_instance)
345 if prov is None:
346 continue
347 if not isinstance(prov, MusicProvider):
348 continue
349 if ProviderFeature.SIMILAR_TRACKS not in prov.supported_features:
350 continue
351 # Grab similar tracks from the music provider
352 return await prov.get_similar_tracks(
353 prov_track_id=prov_mapping.item_id, limit=limit
354 )
355
356 if not allow_lookup:
357 return []
358
359 # check if we have any provider that supports dynamic tracks
360 # TODO: query metadata provider(s) (such as lastfm?)
361 # to get similar tracks (or tracks from similar artists)
362 music_prov: MusicProvider | None = None
363 for prov in self.mass.music.providers:
364 if ProviderFeature.SIMILAR_TRACKS in prov.supported_features:
365 music_prov = prov
366 break
367 if music_prov is None:
368 msg = "No Music Provider found that supports requesting similar tracks."
369 raise UnsupportedFeaturedException(msg)
370
371 if mappings := await self.match_provider(ref_item, music_prov):
372 if ref_item.provider == "library":
373 # update database with new provider mappings
374 await self.add_provider_mappings(ref_item.item_id, mappings)
375 ref_item.provider_mappings.update(mappings)
376 return await music_prov.get_similar_tracks(
377 prov_track_id=mappings[0].item_id, limit=limit
378 )
379
380 return []
381
382 async def remove_item_from_library(self, item_id: str | int, recursive: bool = True) -> None:
383 """Delete record from the database."""
384 db_id = int(item_id) # ensure integer
385 # delete entry(s) from albumtracks table
386 await self.mass.music.database.delete(DB_TABLE_ALBUM_TRACKS, {"track_id": db_id})
387 # delete entry(s) from trackartists table
388 await self.mass.music.database.delete(DB_TABLE_TRACK_ARTISTS, {"track_id": db_id})
389 # delete the track itself from db
390 await super().remove_item_from_library(db_id)
391
392 async def get_preview_url(self, provider_instance_id_or_domain: str, item_id: str) -> str:
393 """Return url to short preview sample."""
394 track = await self.get_provider_item(item_id, provider_instance_id_or_domain)
395 # prefer provider-provided preview
396 if preview := track.metadata.preview:
397 return preview
398 # fallback to a preview/sample hosted by our own webserver
399 enc_track_id = urllib.parse.quote(item_id)
400 return (
401 f"{self.mass.webserver.base_url}/preview?"
402 f"provider={provider_instance_id_or_domain}&item_id={enc_track_id}"
403 )
404
405 async def get_library_track_albums(
406 self,
407 item_id: str | int,
408 ) -> list[Album]:
409 """Return all in-library albums for a track."""
410 db_id = int(item_id) # ensure integer
411 subquery = (
412 f"SELECT album_id FROM {DB_TABLE_ALBUM_TRACKS} "
413 f"WHERE {DB_TABLE_ALBUM_TRACKS}.track_id = :track_id"
414 )
415 query = f"{DB_TABLE_ALBUMS}.item_id in ({subquery})"
416 return await self.mass.music.albums.get_library_items_by_query(
417 extra_query_parts=[query],
418 extra_query_params={"track_id": db_id},
419 )
420
421 async def match_provider(
422 self,
423 base_track: Track,
424 provider: MusicProvider,
425 strict: bool = True,
426 ref_albums: list[Album] | None = None,
427 ) -> list[ProviderMapping]:
428 """
429 Try to find match on (streaming) provider for the provided track.
430
431 This is used to link objects of different providers/qualities together.
432 """
433 if ref_albums is None:
434 ref_albums = await self.albums(base_track.item_id, base_track.provider)
435 self.logger.debug("Trying to match track %s on provider %s", base_track.name, provider.name)
436 matches: list[ProviderMapping] = []
437 for artist in base_track.artists:
438 if matches:
439 break
440 search_str = f"{artist.name} - {base_track.name}"
441 search_result = await self.search(search_str, provider.domain)
442 for search_result_item in search_result:
443 if not search_result_item.available:
444 continue
445 # do a basic compare first
446 if not compare_media_item(base_track, search_result_item, strict=False):
447 continue
448 # we must fetch the full version, search results can be simplified objects
449 prov_track = await self.get_provider_item(
450 search_result_item.item_id,
451 search_result_item.provider,
452 fallback=search_result_item,
453 )
454 if compare_track(base_track, prov_track, strict=strict, track_albums=ref_albums):
455 matches.extend(search_result_item.provider_mappings)
456
457 if not matches:
458 self.logger.debug(
459 "Could not find match for Track %s on provider %s",
460 base_track.name,
461 provider.name,
462 )
463 return matches
464
465 async def match_providers(self, db_track: Track) -> None:
466 """
467 Try to find matching track on all providers for the provided (database) track_id.
468
469 This is used to link objects of different providers/qualities together.
470 """
471 if db_track.provider != "library":
472 return # Matching only supported for database items
473
474 track_albums = await self.albums(db_track.item_id, db_track.provider)
475 # try to find match on all providers
476 processed_domains = set()
477 for provider in self.mass.music.providers:
478 if provider.domain in processed_domains:
479 continue
480 if ProviderFeature.SEARCH not in provider.supported_features:
481 continue
482 if not provider.library_supported(MediaType.TRACK):
483 continue
484 if not provider.is_streaming_provider:
485 # matching on unique providers is pointless as they push (all) their content to MA
486 continue
487 if match := await self.match_provider(
488 db_track, provider, strict=True, ref_albums=track_albums
489 ):
490 # 100% match, we update the db with the additional provider mapping(s)
491 await self.add_provider_mappings(db_track.item_id, match)
492 processed_domains.add(provider.domain)
493
494 async def radio_mode_base_tracks(
495 self,
496 item: Track,
497 preferred_provider_instances: list[str] | None = None,
498 ) -> list[Track]:
499 """
500 Get the list of base tracks from the controller used to calculate the dynamic radio.
501
502 :param item: The Track to get base tracks for.
503 :param preferred_provider_instances: List of preferred provider instance IDs to use.
504 """
505 return [item]
506
507 async def _add_library_item(self, item: Track, overwrite_existing: bool = False) -> int:
508 """Add a new item record to the database."""
509 if not isinstance(item, Track): # TODO: Remove this once the codebase is fully typed
510 msg = "Not a valid Track object (ItemMapping can not be added to db)" # type: ignore[unreachable]
511 raise InvalidDataError(msg)
512 if not item.artists:
513 msg = "Track is missing artist(s)"
514 raise InvalidDataError(msg)
515 db_id = await self.mass.music.database.insert(
516 self.db_table,
517 {
518 "name": item.name,
519 "sort_name": item.sort_name,
520 "version": item.version,
521 "duration": item.duration,
522 "favorite": item.favorite,
523 "external_ids": serialize_to_json(item.external_ids),
524 "metadata": serialize_to_json(item.metadata),
525 "search_name": create_safe_string(item.name, True, True),
526 "search_sort_name": create_safe_string(item.sort_name or "", True, True),
527 "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
528 },
529 )
530 # update/set provider_mappings table
531 await self.set_provider_mappings(db_id, item.provider_mappings)
532 # set track artist(s)
533 await self._set_track_artists(db_id, item.artists)
534 # handle track album
535 if item.album:
536 await self._set_track_album(
537 db_id=db_id,
538 album=item.album,
539 disc_number=getattr(item, "disc_number", 0),
540 track_number=getattr(item, "track_number", 0),
541 )
542 self.logger.debug("added %s to database (id: %s)", item.name, db_id)
543 return db_id
544
545 async def _update_library_item(
546 self, item_id: str | int, update: Track, overwrite: bool = False
547 ) -> None:
548 """Update Track record in the database, merging data."""
549 db_id = int(item_id) # ensure integer
550 cur_item = await self.get_library_item(db_id)
551 metadata = update.metadata if overwrite else cur_item.metadata.update(update.metadata)
552 cur_item.external_ids.update(update.external_ids)
553 name = update.name if overwrite else cur_item.name
554 sort_name = update.sort_name if overwrite else cur_item.sort_name or update.sort_name
555 await self.mass.music.database.update(
556 self.db_table,
557 {"item_id": db_id},
558 {
559 "name": name,
560 "sort_name": sort_name,
561 "version": update.version if overwrite else cur_item.version or update.version,
562 "duration": update.duration if overwrite else cur_item.duration or update.duration,
563 "metadata": serialize_to_json(metadata),
564 "external_ids": serialize_to_json(
565 update.external_ids if overwrite else cur_item.external_ids
566 ),
567 "search_name": create_safe_string(name, True, True),
568 "search_sort_name": create_safe_string(sort_name or "", True, True),
569 "timestamp_added": int(update.date_added.timestamp())
570 if update.date_added
571 else UNSET,
572 },
573 )
574 # update/set provider_mappings table
575 provider_mappings = (
576 update.provider_mappings
577 if overwrite
578 else {*update.provider_mappings, *cur_item.provider_mappings}
579 )
580 await self.set_provider_mappings(db_id, provider_mappings, overwrite)
581 # set track artist(s)
582 artists = update.artists if overwrite else cur_item.artists + update.artists
583 await self._set_track_artists(db_id, artists, overwrite=overwrite)
584 # update/set track album
585 if update.album:
586 await self._set_track_album(
587 db_id=db_id,
588 album=update.album,
589 disc_number=update.disc_number or cur_item.disc_number,
590 track_number=update.track_number or cur_item.track_number,
591 overwrite=overwrite,
592 )
593 self.logger.debug("updated %s in database: (id %s)", update.name, db_id)
594
595 async def _set_track_album(
596 self,
597 db_id: int,
598 album: Album | ItemMapping,
599 disc_number: int,
600 track_number: int,
601 overwrite: bool = False,
602 ) -> None:
603 """
604 Store Track Album info.
605
606 A track can exist on multiple albums so we have a mapping table between
607 albums and tracks which stores the relation between the two and it also
608 stores the track and disc number of the track within an album.
609 For digital releases, the discnumber will be just 0 or 1.
610 Track number should start counting at 1.
611 """
612 db_album: Album | ItemMapping | None = None
613 if album.provider == "library":
614 db_album = album
615 elif existing := await self.mass.music.albums.get_library_item_by_prov_id(
616 album.item_id, album.provider
617 ):
618 db_album = existing
619
620 if not db_album or overwrite:
621 # ensure we have an actual album object
622 if isinstance(album, ItemMapping):
623 db_album = await self.mass.music.albums.add_item_mapping_as_album_to_library(album)
624 else:
625 db_album = await self.mass.music.albums.add_item_to_library(
626 album,
627 overwrite_existing=overwrite,
628 )
629 # write (or update) record in album_tracks table
630 await self.mass.music.database.insert_or_replace(
631 DB_TABLE_ALBUM_TRACKS,
632 {
633 "track_id": db_id,
634 "album_id": int(db_album.item_id),
635 "disc_number": disc_number,
636 "track_number": track_number,
637 },
638 )
639
640 async def _set_track_artists(
641 self,
642 db_id: int,
643 artists: Iterable[Artist | ItemMapping],
644 overwrite: bool = False,
645 ) -> None:
646 """Store Track Artists."""
647 if overwrite:
648 # on overwrite, clear the track_artists table first
649 await self.mass.music.database.delete(
650 DB_TABLE_TRACK_ARTISTS,
651 {
652 "track_id": db_id,
653 },
654 )
655 artist_mappings: UniqueList[ItemMapping] = UniqueList()
656 for artist in artists:
657 mapping = await self._set_track_artist(db_id, artist=artist, overwrite=overwrite)
658 artist_mappings.append(mapping)
659
660 async def _set_track_artist(
661 self, db_id: int, artist: Artist | ItemMapping, overwrite: bool = False
662 ) -> ItemMapping:
663 """Store Track Artist info."""
664 db_artist: Artist | ItemMapping | None = None
665 if artist.provider == "library":
666 db_artist = artist
667 elif existing := await self.mass.music.artists.get_library_item_by_prov_id(
668 artist.item_id, artist.provider
669 ):
670 db_artist = existing
671
672 if not db_artist or overwrite:
673 # Convert ItemMapping to Artist if needed
674 artist_to_add = (
675 self.mass.music.artists.artist_from_item_mapping(artist)
676 if isinstance(artist, ItemMapping)
677 else artist
678 )
679 db_artist = await self.mass.music.artists.add_item_to_library(
680 artist_to_add, overwrite_existing=overwrite
681 )
682 # write (or update) record in track_artists table
683 await self.mass.music.database.insert_or_replace(
684 DB_TABLE_TRACK_ARTISTS,
685 {
686 "track_id": db_id,
687 "artist_id": int(db_artist.item_id),
688 },
689 )
690 return ItemMapping.from_item(db_artist)
691