@property
def providers(self) -> list[MusicProvider]:
- """Return all loaded/running MusicProviders (instances)."""
- return self.mass.get_providers(ProviderType.MUSIC)
+ """
+ Return all loaded/running MusicProviders (instances).
+
+ Note that this applies user provider filters (for all user types).
+ """
+ user = get_current_user()
+ user_provider_filter = user.provider_filter if user else None
+ return [
+ x
+ for x in self.mass.providers
+ if x.type == ProviderType.MUSIC
+ and (not user_provider_filter or x.instance_id in user_provider_filter)
+ ]
@api_command("music/sync")
async def start_sync(
:param media_types: A list of media_types to include.
:param limit: number of items to return in the search (per type).
"""
- # use a (short-lived) cache to avoid repeated searches
- cache_key = f"{search_query}{'-'.join(sorted([mt.value for mt in media_types]))}-{limit}-{library_only}" # noqa: E501
- if user := get_current_user():
- cache_key += user.user_id
+ # use cache to avoid repeated searches
+ search_providers = sorted(self.get_unique_providers())
+ cache_provider_key = "library" if library_only else ",".join(search_providers)
+ cache_key = f"{search_query}{'-'.join(sorted([mt.value for mt in media_types]))}-{limit}-{library_only}-{cache_provider_key}" # noqa: E501
if cache := await self.mass.cache.get(
key=cache_key, provider=self.domain, category=CACHE_CATEGORY_SEARCH_RESULTS
):
for prov_mapping in item.provider_mappings
}
# include results from library + all (unique) music providers
- search_providers = self.get_unique_providers()
results_per_provider += await asyncio.gather(
*[
self._search_provider(
await self.mass.cache.set(
key=cache_key,
data=result,
- expiration=300,
+ expiration=600,
provider=self.domain,
category=CACHE_CATEGORY_SEARCH_RESULTS,
)
def get_provider_instances(
self, domain: str, return_unavailable: bool = False
) -> list[MusicProvider]:
- """Return all provider instances for a given domain."""
- return [
- prov
- # don't use self.providers here as that applies user filters
- for prov in self.mass.providers
- if isinstance(prov, MusicProvider)
- and prov.domain == domain
- and (return_unavailable or prov.available)
- ]
+ """
+ Return all provider instances for a given domain.
+
+ Note that this skips user filters so may only be called from internal code.
+ """
+ return cast(
+ "list[MusicProvider]",
+ self.mass.get_provider_instances(domain, return_unavailable, ProviderType.MUSIC),
+ )
- def get_unique_providers(self) -> set[str]:
+ def get_unique_providers(self) -> list[str]:
"""
Return all unique MusicProvider (instance or domain) ids.
This will return a set of provider instance ids but will only return
a single instance_id per streaming provider domain.
+
+ Applies user provider filters (for non-admin users).
"""
processed_domains: set[str] = set()
# Get user provider filter if set
user = get_current_user()
user_provider_filter = user.provider_filter if user and user.provider_filter else None
- result: set[str] = set()
+ result: list[str] = []
for provider in self.providers:
if provider.is_streaming_provider and provider.domain in processed_domains:
continue
if user_provider_filter and provider.instance_id not in user_provider_filter:
continue
- result.add(provider.instance_id)
+ result.append(provider.instance_id)
processed_domains.add(provider.domain)
return result
from contextlib import suppress
from typing import TYPE_CHECKING, Any, Concatenate, TypedDict, cast, overload
+from music_assistant_models.auth import UserRole
from music_assistant_models.constants import (
PLAYER_CONTROL_FAKE,
PLAYER_CONTROL_NATIVE,
"""
Return all registered players.
+ Note that this applies user filters for players (for non admin users).
+
:param return_unavailable [bool]: Include unavailable players.
:param return_disabled [bool]: Include disabled players.
:param provider_filter [str]: Optional filter by provider lookup key.
:return: List of Player objects.
"""
current_user = get_current_user()
- user_filter = current_user.player_filter if current_user else []
+ user_filter = (
+ current_user.player_filter
+ if current_user and current_user.role != UserRole.ADMIN
+ else None
+ )
return [
player
for player in self._players.values()
:raises PlayerUnavailableError: If player is unavailable and raise_unavailable is True.
:return: Player object or None.
"""
+ current_user = get_current_user()
+ user_filter = (
+ current_user.player_filter
+ if current_user and current_user.role != UserRole.ADMIN
+ else None
+ )
+ if current_user and user_filter and player_id not in user_filter:
+ msg = f"{current_user.username} does not have access to player {player_id}"
+ raise InsufficientPermissions(msg)
if player := self.get(player_id, raise_unavailable):
return player.state
return None
:param name: Name of the player.
:return: PlayerState object or None.
"""
+ current_user = get_current_user()
+ user_filter = (
+ current_user.player_filter
+ if current_user and current_user.role != UserRole.ADMIN
+ else None
+ )
if player := self.get_player_by_name(name):
+ if current_user and user_filter and player.player_id not in user_filter:
+ msg = f"{current_user.username} does not have access to player {player.player_id}"
+ raise InsufficientPermissions(msg)
return player.state
return None
import aiofiles
from aiofiles.os import wrap
from music_assistant_models.api import ServerInfoMessage
+from music_assistant_models.auth import UserRole
from music_assistant_models.enums import EventType, ProviderType
from music_assistant_models.errors import MusicAssistantError, SetupFailedError
from music_assistant_models.event import MassEvent
def get_providers(
self, provider_type: ProviderType | None = None
) -> list[ProviderInstanceType]:
- """Return all loaded/running Providers (instances), optionally filtered by ProviderType."""
- user = get_current_user()
- user_provider_filter = user.provider_filter if user else None
+ """
+ Return all loaded/running Providers (instances).
+ Optionally filtered by ProviderType.
+ Note that this applies user filters for music providers (for non admin users).
+ """
+ user = get_current_user()
+ user_provider_filter = (
+ user.provider_filter if user and user.role != UserRole.ADMIN else None
+ )
return [
x
for x in self._providers.values()
if (provider_type is None or provider_type == x.type)
- # handle optional user (music) provider filter
+ # apply user provider filter
and (
not user_provider_filter
or x.instance_id in user_provider_filter
)
]
- @api_command("logging/get")
+ @api_command("logging/get", required_role=UserRole.ADMIN)
async def get_application_log(self) -> str:
"""Return the application log from file."""
logfile = os.path.join(self.storage_path, "musicassistant.log")
@property
def providers(self) -> list[ProviderInstanceType]:
- """Return all loaded/running Providers (instances)."""
+ """
+ Return all loaded/running Providers (instances).
+
+ Note that this skips user filters so may only be called from internal code.
+ """
return list(self._providers.values())
@overload
return prov
return None
+ def get_provider_instances(
+ self,
+ domain: str,
+ return_unavailable: bool = False,
+ provider_type: ProviderType | None = None,
+ ) -> list[ProviderInstanceType]:
+ """
+ Return all provider instances for a given domain.
+
+ Note that this skips user filters so may only be called from internal code.
+ """
+ return [
+ prov
+ for prov in self._providers.values()
+ if (provider_type is None or provider_type == prov.type)
+ and prov.domain == domain
+ and (return_unavailable or prov.available)
+ ]
+
def signal_event(
self,
event: EventType,
projects / music-assistant-server.git / commitdiff