ProviderConfig,
)
from music_assistant.common.models.enums import EventType, ProviderType
-from music_assistant.common.models.errors import (
- InvalidDataError,
- PlayerUnavailableError,
- ProviderUnavailableError,
-)
+from music_assistant.common.models.errors import InvalidDataError, PlayerUnavailableError
from music_assistant.constants import CONF_PLAYERS, CONF_PROVIDERS, CONF_SERVER_ID, ENCRYPT_SUFFIX
from music_assistant.server.helpers.api import api_command
from music_assistant.server.models.player_provider import PlayerProvider
"""Update ProviderConfig."""
config = self.get_provider_config(instance_id)
changed_keys = config.update(update)
- try:
- prov = self.mass.get_provider(instance_id)
- available = prov.available
- except ProviderUnavailableError:
- available = False
+ available = prov.available if (prov := self.mass.get_provider(instance_id)) else False
if not changed_keys and (config.enabled == available):
# no changes
return
def get_player_config(self, player_id: str) -> PlayerConfig:
"""Return configuration for a single player."""
if raw_conf := self.get(f"{CONF_PLAYERS}/{player_id}"):
- try:
- prov = self.mass.get_provider(raw_conf["provider"])
+ if prov := self.mass.get_provider(raw_conf["provider"]):
prov_entries = prov.get_player_config_entries(player_id)
- except (ProviderUnavailableError, PlayerUnavailableError):
+ else:
prov_entries = tuple()
raw_conf["available"] = False
raw_conf["name"] = (
raw_conf.get("name") or raw_conf.get("default_name") or raw_conf["player_id"]
)
-
entries = DEFAULT_PLAYER_CONFIG_ENTRIES + prov_entries
return PlayerConfig.parse(entries, raw_conf)
raise KeyError(f"No config found for player id {player_id}")
from music_assistant.common.helpers.json import serialize_to_json
from music_assistant.common.models.enums import EventType, ProviderFeature
-from music_assistant.common.models.errors import (
- MediaNotFoundError,
- ProviderUnavailableError,
- UnsupportedFeaturedException,
-)
+from music_assistant.common.models.errors import MediaNotFoundError, UnsupportedFeaturedException
from music_assistant.common.models.media_items import (
Album,
AlbumType,
provider_instance: str | None = None,
) -> list[Track]:
"""Return album tracks for the given provider album id."""
- try:
- prov = self.mass.get_provider(provider_instance or provider_domain)
- except ProviderUnavailableError:
+ prov = self.mass.get_provider(provider_instance or provider_domain)
+ if prov is None:
return []
full_album = await self.get_provider_item(item_id, provider_instance or provider_domain)
limit: int = 25,
):
"""Generate a dynamic list of tracks based on the album content."""
- try:
- prov = self.mass.get_provider(provider_instance or provider_domain)
- except ProviderUnavailableError:
+ prov = self.mass.get_provider(provider_instance or provider_domain)
+ if prov is None:
return []
if ProviderFeature.SIMILAR_TRACKS not in prov.supported_features:
return []
from music_assistant.common.helpers.json import serialize_to_json
from music_assistant.common.models.enums import EventType, ProviderFeature
-from music_assistant.common.models.errors import (
- MediaNotFoundError,
- ProviderUnavailableError,
- UnsupportedFeaturedException,
-)
+from music_assistant.common.models.errors import MediaNotFoundError, UnsupportedFeaturedException
from music_assistant.common.models.media_items import (
Album,
AlbumType,
cache_checksum: Any = None,
) -> list[Track]:
"""Return top tracks for an artist on given provider."""
- try:
- prov = self.mass.get_provider(provider_instance or provider_domain)
- except ProviderUnavailableError:
+ prov = self.mass.get_provider(provider_instance or provider_domain)
+ if prov is None:
return []
# prefer cache items (if any)
cache_key = f"{prov.instance_id}.artist_toptracks.{item_id}"
cache_checksum: Any = None,
) -> list[Album]:
"""Return albums for an artist on given provider."""
- try:
- prov = self.mass.get_provider(provider_instance or provider_domain)
- except ProviderUnavailableError:
+ prov = self.mass.get_provider(provider_instance or provider_domain)
+ if prov is None:
return []
# prefer cache items (if any)
cache_key = f"{prov.instance_id}.artist_albums.{item_id}"
limit: int = 25,
):
"""Generate a dynamic list of tracks based on the artist's top tracks."""
- try:
- prov = self.mass.get_provider(provider_instance or provider_domain)
- except ProviderUnavailableError:
+ prov = self.mass.get_provider(provider_instance or provider_domain)
+ if prov is None:
return []
if ProviderFeature.SIMILAR_TRACKS not in prov.supported_features:
return []
from music_assistant.common.helpers.json import serialize_to_json
from music_assistant.common.models.enums import EventType, MediaType, ProviderFeature
-from music_assistant.common.models.errors import MediaNotFoundError, ProviderUnavailableError
+from music_assistant.common.models.errors import MediaNotFoundError
from music_assistant.common.models.media_items import (
MediaItemType,
PagedItems,
self.item_cls.from_db_row(db_row)
for db_row in await self.mass.music.database.search(self.db_table, search_query)
]
-
- try:
- prov = self.mass.get_provider(provider_instance or provider_domain)
- except ProviderUnavailableError:
+ prov = self.mass.get_provider(provider_instance or provider_domain)
+ if prov is None:
return []
if ProviderFeature.SEARCH not in prov.supported_features:
return []
item = await self.get_db_item(item_id)
else:
provider = self.mass.get_provider(provider_domain_or_instance_id)
- item = await provider.get_item(self.media_type, item_id)
+ item = (await provider.get_item(self.media_type, item_id)) if provider else None
if not item:
raise MediaNotFoundError(
f"{self.media_type.value}://{item_id} not found on provider {provider_domain_or_instance_id}" # noqa: E501
"""Return a dynamic list of tracks based on the given item."""
ref_item = await self.get(item_id, provider_domain, provider_instance)
for prov_mapping in ref_item.provider_mappings:
- try:
- prov = self.mass.get_provider(prov_mapping.provider_instance)
- except ProviderUnavailableError:
- continue
- if not prov.available:
+ prov = self.mass.get_provider(prov_mapping.provider_instance)
+ if prov is None:
continue
if ProviderFeature.SIMILAR_TRACKS not in prov.supported_features:
continue
),
None,
)
- if provider is None:
- raise ProviderUnavailableError(
- "No provider available which allows playlists creation."
- )
+ if provider is None:
+ raise ProviderUnavailableError("No provider available which allows playlists creation.")
return await provider.create_playlist(name)
from music_assistant.common.helpers.json import serialize_to_json
from music_assistant.common.models.enums import EventType, MediaType, ProviderFeature
-from music_assistant.common.models.errors import (
- MediaNotFoundError,
- ProviderUnavailableError,
- UnsupportedFeaturedException,
-)
+from music_assistant.common.models.errors import MediaNotFoundError, UnsupportedFeaturedException
from music_assistant.common.models.media_items import (
Album,
Artist,
limit: int = 25,
):
"""Generate a dynamic list of tracks based on the track."""
- try:
- prov = self.mass.get_provider(provider_instance or provider_domain)
- except ProviderUnavailableError:
+ prov = self.mass.get_provider(provider_instance or provider_domain)
+ if prov is None:
return []
if ProviderFeature.SIMILAR_TRACKS not in prov.supported_features:
return []
from __future__ import annotations
import asyncio
-import contextlib
import logging
import os
import urllib.parse
from base64 import b64encode
+from contextlib import suppress
from random import shuffle
from time import time
from typing import TYPE_CHECKING
from aiohttp import web
from music_assistant.common.models.enums import ImageType, MediaType, ProviderFeature, ProviderType
-from music_assistant.common.models.errors import ProviderUnavailableError
+from music_assistant.common.models.errors import MediaNotFoundError
from music_assistant.common.models.media_items import (
Album,
Artist,
if artist.image:
continue
# simply grabbing the full artist will trigger a full fetch
- with contextlib.suppress(ProviderUnavailableError):
+ with suppress(MediaNotFoundError):
await self.mass.music.artists.get(artist.item_id, artist.provider, lazy=False)
# this is slow on purpose to not cause stress on the metadata providers
await asyncio.sleep(30)
from music_assistant.common.helpers.datetime import utc_timestamp
from music_assistant.common.helpers.uri import parse_uri
from music_assistant.common.models.enums import EventType, MediaType, ProviderFeature, ProviderType
-from music_assistant.common.models.errors import MusicAssistantError, ProviderUnavailableError
+from music_assistant.common.models.errors import MusicAssistantError
from music_assistant.common.models.media_items import (
BrowseFolder,
MediaItem,
:param limit: number of items to return in the search (per type).
"""
assert provider_domain or provider_instance, "Provider needs to be supplied"
- try:
- prov = self.mass.get_provider(provider_instance or provider_domain)
- except ProviderUnavailableError:
+ prov = self.mass.get_provider(provider_instance or provider_domain)
+ if not prov:
return SearchResults()
if ProviderFeature.SEARCH not in prov.supported_features:
return SearchResults()
"%s got new media_status for group: %s", self.castplayer.player.display_name, group_uuid
)
+ def load_media_failed(self, item, error_code):
+ """Call when media failed to load."""
+ self.prov.logger.warning("Load media failed: %s - error code: %s", item, error_code)
+
def invalidate(self):
"""
Invalidate this status listener.
from music_assistant.common.helpers.util import get_ip, get_ip_pton, select_free_port
from music_assistant.common.models.config_entries import ProviderConfig
from music_assistant.common.models.enums import EventType, ProviderType
-from music_assistant.common.models.errors import ProviderUnavailableError, SetupFailedError
+from music_assistant.common.models.errors import SetupFailedError
from music_assistant.common.models.event import MassEvent
from music_assistant.common.models.provider import ProviderManifest
from music_assistant.constants import CONF_PROVIDERS, CONF_SERVER_ID, CONF_WEB_IP, ROOT_LOGGER_NAME
"""Return all loaded/running Providers (instances)."""
return list(self._providers.values())
- def get_provider(self, provider_instance_or_domain: str) -> ProviderInstanceType:
+ def get_provider(
+ self, provider_instance_or_domain: str, return_unavailable: bool = False
+ ) -> ProviderInstanceType | None:
"""Return provider by instance id (or domain)."""
- if prov := self._providers.get(provider_instance_or_domain):
+ prov = self._providers.get(provider_instance_or_domain)
+ if prov is not None and (return_unavailable or prov.available):
return prov
for prov in self._providers.values():
- if prov.domain == provider_instance_or_domain:
+ if prov.domain == provider_instance_or_domain and return_unavailable or prov.available:
return prov
- raise ProviderUnavailableError(f"Provider {provider_instance_or_domain} is not available")
+ LOGGER.warning("Provider {provider_instance_or_domain} is not available")
+ return None
def signal_event(
self,
# handle dependency on other provider
if prov_manifest.depends_on:
for _ in range(30):
- try:
- self.get_provider(prov_manifest.depends_on)
+ if self.get_provider(prov_manifest.depends_on):
break
- except ProviderUnavailableError:
- await asyncio.sleep(1)
+ await asyncio.sleep(1)
else:
raise SetupFailedError(
f"Provider {domain} depends on {prov_manifest.depends_on} "
projects / music-assistant-server.git / commitdiff