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