album.artist.provider,
lazy=True,
details=album.artist,
+ add_to_db=add_to_db,
)
return album
) -> list[Album]:
"""Return all versions of an album we can find on all providers."""
assert provider_domain or provider_instance, "Provider type or ID must be specified"
- album = await self.get(item_id, provider_domain or provider_instance)
+ album = await self.get(item_id, provider_domain or provider_instance, add_to_db=False)
# perform a search on all provider(types) to collect all versions/variants
provider_domains = {item.domain for item in self.mass.music.providers}
search_query = f"{album.artist.name} - {album.name}"
for track in await self._get_provider_album_tracks(
prov_mapping.item_id, prov_mapping.provider_instance
):
- await self.mass.music.tracks.get(track.item_id, track.provider, details=track)
+ await self.mass.music.tracks.get(
+ track.item_id, track.provider, details=track, add_to_db=True
+ )
self.mass.signal_event(
EventType.MEDIA_ITEM_UPDATED if existing else EventType.MEDIA_ITEM_ADDED,
db_item.uri,
) -> list[Track]:
"""Return top tracks for an artist."""
if not artist:
- artist = await self.get(item_id, provider_domain, provider_instance)
+ artist = await self.get(item_id, provider_domain, provider_instance, add_to_db=False)
# get results from all providers
coros = [
self.get_provider_artist_toptracks(
) -> list[Album]:
"""Return (all/most popular) albums for an artist."""
if not artist:
- artist = await self.get(item_id, provider_domain or provider_instance)
+ artist = await self.get(item_id, provider_domain or provider_instance, add_to_db=False)
# get results from all providers
coros = [
self.get_provider_artist_albums(
assert (
provider_domain or provider_instance
), "provider_domain or provider_instance must be supplied"
- if not add_to_db:
+ if not add_to_db and "database" in (provider_domain, provider_instance):
return await self.get_provider_item(item_id, provider_instance or provider_domain)
if details and details.provider == "database":
details = None
if db_item and (time() - (db_item.metadata.last_refresh or 0)) > REFRESH_INTERVAL:
# it's been too long since the full metadata was last retrieved (or never at all)
force_refresh = True
- if db_item and force_refresh:
+ if db_item and force_refresh and add_to_db:
# get (first) provider item id belonging to this db item
provider_instance, item_id = await self.get_provider_mapping(db_item)
elif db_item:
if not details:
# we couldn't get a match from any of the providers, raise error
raise MediaNotFoundError(f"Item not found: {provider_domain or id}/{item_id}")
+ if not add_to_db:
+ return details
# create task to add the item to the db, including matching metadata etc. takes some time
# in 99% of the cases we just return lazy because we want the details as fast as possible
# only if we really need to wait for the result (e.g. to prevent race conditions), we
self.mass.register_api_command("music/tracks", self.db_items)
self.mass.register_api_command("music/track", self.get)
self.mass.register_api_command("music/track/versions", self.versions)
+ self.mass.register_api_command("music/track/albums", self.albums)
self.mass.register_api_command("music/track/update", self.update_db_item)
self.mass.register_api_command("music/track/delete", self.delete_db_item)
self.mass.register_api_command("music/track/preview", self.get_preview_url)
track.album.provider,
lazy=True,
details=track.album,
+ add_to_db=add_to_db,
)
except MediaNotFoundError:
# edge case where playlist track has invalid albumdetails
for artist in track.artists:
full_artists.append(
await self.mass.music.artists.get(
- artist.item_id, artist.provider, lazy=True, details=artist
+ artist.item_id, artist.provider, lazy=True, details=artist, add_to_db=add_to_db
)
)
track.artists = full_artists
) -> list[Track]:
"""Return all versions of a track we can find on all providers."""
assert provider_domain or provider_instance, "Provider type or ID must be specified"
- track = await self.get(item_id, provider_domain or provider_instance)
+ track = await self.get(item_id, provider_domain or provider_instance, add_to_db=False)
# perform a search on all provider(types) to collect all versions/variants
provider_domains = {prov.domain for prov in self.mass.music.providers}
search_query = f"{track.artist.name} - {track.name}"
# return the aggregated result
return all_versions.values()
+ async def albums(
+ self,
+ item_id: str,
+ provider_domain: str | None = None,
+ provider_instance: str | None = None,
+ ) -> list[Album]:
+ """Return all albums the track appears on."""
+ assert provider_domain or provider_instance, "Provider type or ID must be specified"
+ track = await self.get(item_id, provider_domain or provider_instance, add_to_db=False)
+ return await asyncio.gather(
+ *[
+ self.mass.music.albums.get(album.item_id, album.provider, add_to_db=False)
+ for album in track.albums
+ ]
+ )
+
async def get_preview_url(self, provider_domain: str, item_id: str) -> str:
"""Return url to short preview sample."""
track = await self.get_provider_item(item_id, provider_domain)
projects / music-assistant-server.git / commitdiff