from uuid import uuid4
import aiofiles
+import shortuuid
from aiofiles.os import wrap
from cryptography.fernet import Fernet, InvalidToken
instance_id = provider_domain
name = manifest.name
else:
- instance_id = f"{provider_domain}{len(existing)+1}"
- name = f"{manifest.name} {len(existing)+1}"
+ random_id = shortuuid.random(6)
+ instance_id = f"{provider_domain}_{random_id}"
+ name = f"{manifest.name} {random_id}"
# all checks passed, create config object
config_entries = await self.get_provider_config_entries(
provider_domain=provider_domain, instance_id=instance_id, values=values
await self.mass.music.albums.remove_item_from_library(album_id)
# check if any artists need to be cleaned up
for artist_id in artist_ids:
- if not self.mass.music.artists.albums(
- artist_id, "library"
- ) and self.mass.music.artists.tracks(artist_id, "library"):
+ artist_albums = await self.mass.music.artists.albums(artist_id, "library")
+ artist_tracks = self.mass.music.artists.tracks(artist_id, "library")
+ if not (artist_albums or artist_tracks):
await self.mass.music.artists.remove_item_from_library(album_id)
async def get_artist(self, prov_artist_id: str) -> Artist:
import aiohttp.client_exceptions
from asyncio_throttle import Throttler
from mashumaro import DataClassDictMixin
+from mashumaro.exceptions import MissingField
from music_assistant.common.helpers.util import parse_title_and_version
from music_assistant.common.models.config_entries import ConfigEntry, ConfigValueType
if "id" not in result:
result["id"] = artist_id
# TODO: Parse all the optional data like relations and such
- return MusicBrainzArtist.from_dict(replace_hyphens(result))
+ try:
+ return MusicBrainzArtist.from_dict(replace_hyphens(result))
+ except MissingField as err:
+ raise InvalidDataError from err
raise InvalidDataError("Invalid MusicBrainz Artist ID provided")
async def get_recording_details(
if result := await self.get_data(f"recording/{recording_id}?inc=artists+releases"):
if "id" not in result:
result["id"] = recording_id
- return MusicBrainzRecording.from_dict(replace_hyphens(result))
+ try:
+ return MusicBrainzRecording.from_dict(replace_hyphens(result))
+ except MissingField as err:
+ raise InvalidDataError from err
raise InvalidDataError("Invalid ISRC provided")
async def get_releasegroup_details(
if result := await self.get_data(endpoint):
if "id" not in result:
result["id"] = releasegroup_id
- return MusicBrainzReleaseGroup.from_dict(replace_hyphens(result))
+ try:
+ return MusicBrainzReleaseGroup.from_dict(replace_hyphens(result))
+ except MissingField as err:
+ raise InvalidDataError from err
raise InvalidDataError("Invalid MusicBrainz ReleaseGroup ID or barcode provided")
async def get_artist_details_by_album(
item_id=str(artist_obj["id"]),
provider_domain=self.domain,
provider_instance=self.instance_id,
- url=artist_obj.get("url", f'https://open.qobuz.com/artist/{artist_obj["id"]}'),
+ url=f'https://open.qobuz.com/artist/{artist_obj["id"]}',
)
},
)
sample_rate=album_obj["maximum_sampling_rate"] * 1000,
bit_depth=album_obj["maximum_bit_depth"],
),
- url=album_obj.get("url", f'https://open.qobuz.com/album/{album_obj["id"]}'),
+ url=f'https://open.qobuz.com/album/{album_obj["id"]}',
)
},
)
sample_rate=track_obj["maximum_sampling_rate"] * 1000,
bit_depth=track_obj["maximum_bit_depth"],
),
- url=track_obj.get("url", f'https://open.qobuz.com/track/{track_obj["id"]}'),
+ url=f'https://open.qobuz.com/track/{track_obj["id"]}',
)
},
**extra_init_kwargs,
item_id=str(playlist_obj["id"]),
provider_domain=self.domain,
provider_instance=self.instance_id,
- url=playlist_obj.get(
- "url", f'https://open.qobuz.com/playlist/{playlist_obj["id"]}'
- ),
+ url=f'https://open.qobuz.com/playlist/{playlist_obj["id"]}',
)
},
)