DEFAULT_PROVIDER_CONFIG_ENTRIES,
ENCRYPT_SUFFIX,
)
-from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
from music_assistant.helpers.api import api_command
from music_assistant.helpers.json import JSON_DECODE_EXCEPTIONS, async_json_dumps, async_json_loads
from music_assistant.helpers.util import load_provider_module
self.save()
- @api_command("config/providers")
+ @api_command("config/providers", required_role="admin")
async def get_provider_configs(
self,
provider_type: ProviderType | None = None,
include_values: bool = False,
) -> list[ProviderConfig]:
"""Return all known provider configurations, optionally filtered by ProviderType."""
- user = get_current_user()
- user_provider_filter = user.provider_filter if user else None
-
raw_values = self.get(CONF_PROVIDERS, {})
prov_entries = {x.domain for x in self.mass.get_provider_manifests()}
return [
and (provider_domain is None or prov_conf["domain"] == provider_domain)
# guard for deleted providers
and prov_conf["domain"] in prov_entries
- # filter by user's provider_filter if set
- and (not user_provider_filter or prov_conf["instance_id"] in user_provider_filter)
]
- @api_command("config/providers/get")
+ @api_command("config/providers/get", required_role="admin")
async def get_provider_config(self, instance_id: str) -> ProviderConfig:
"""Return configuration for a single provider."""
if raw_conf := self.get(f"{CONF_PROVIDERS}/{instance_id}", {}):
limit: int = 500,
offset: int = 0,
order_by: str = "sort_name",
- provider: str | None = None,
+ provider: str | list[str] | None = None,
extra_query: str | None = None,
extra_query_params: dict[str, Any] | None = None,
album_types: list[AlbumType] | None = None,
) -> list[Album]:
- """Get in-database albums."""
+ """Get in-database albums.
+
+ :param favorite: Filter by favorite status.
+ :param search: Filter by search query.
+ :param limit: Maximum number of items to return.
+ :param offset: Number of items to skip.
+ :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
+ :param provider: Filter by provider instance ID or domain (single string or list).
+ :param extra_query: Additional SQL query string.
+ :param extra_query_params: Additional query parameters.
+ :param album_types: Filter by album types.
+ """
extra_query_params = extra_query_params or {}
extra_query_parts: list[str] = [extra_query] if extra_query else []
extra_join_parts: list[str] = []
limit: int = 500,
offset: int = 0,
order_by: str = "sort_name",
- provider: str | None = None,
+ provider: str | list[str] | None = None,
extra_query: str | None = None,
extra_query_params: dict[str, Any] | None = None,
album_artists_only: bool = False,
) -> list[Artist]:
- """Get in-database (album) artists."""
+ """Get in-database (album) artists.
+
+ :param favorite: Filter by favorite status.
+ :param search: Filter by search query.
+ :param limit: Maximum number of items to return.
+ :param offset: Number of items to skip.
+ :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
+ :param provider: Filter by provider instance ID or domain (single string or list).
+ :param extra_query: Additional SQL query string.
+ :param extra_query_params: Additional query parameters.
+ :param album_artists_only: Only return artists that have albums.
+ """
extra_query_params = extra_query_params or {}
extra_query_parts: list[str] = [extra_query] if extra_query else []
if album_artists_only:
limit: int = 500,
offset: int = 0,
order_by: str = "sort_name",
- provider: str | None = None,
+ provider: str | list[str] | None = None,
extra_query: str | None = None,
extra_query_params: dict[str, Any] | None = None,
) -> list[Audiobook]:
- """Get in-database audiobooks."""
+ """Get in-database audiobooks.
+
+ :param favorite: Filter by favorite status.
+ :param search: Filter by search query.
+ :param limit: Maximum number of items to return.
+ :param offset: Number of items to skip.
+ :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
+ :param provider: Filter by provider instance ID or domain (single string or list).
+ :param extra_query: Additional SQL query string.
+ :param extra_query_params: Additional query parameters.
+ """
extra_query_params = extra_query_params or {}
extra_query_parts: list[str] = [extra_query] if extra_query else []
result = await self._get_library_items_by_query(
from music_assistant_models.media_items import ItemMapping, MediaItemType, ProviderMapping, Track
from music_assistant.constants import DB_TABLE_PLAYLOG, DB_TABLE_PROVIDER_MAPPINGS, MASS_LOGGER_NAME
+from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
from music_assistant.helpers.compare import compare_media_item, create_safe_string
from music_assistant.helpers.json import json_loads, serialize_to_json
from music_assistant.helpers.util import guard_single_request
limit: int = 500,
offset: int = 0,
order_by: str = "sort_name",
- provider: str | None = None,
+ provider: str | list[str] | None = None,
extra_query: str | None = None,
extra_query_params: dict[str, Any] | None = None,
) -> list[ItemCls]:
favorite: bool | None = None,
search: str | None = None,
order_by: str = "sort_name",
- provider: str | None = None,
+ provider: str | list[str] | None = None,
extra_query: str | None = None,
extra_query_params: dict[str, Any] | None = None,
) -> AsyncGenerator[ItemCls, None]:
limit: int = 500,
offset: int = 0,
order_by: str | None = None,
- provider: str | None = None,
+ provider: str | list[str] | None = None,
extra_query_parts: list[str] | None = None,
extra_query_params: dict[str, Any] | None = None,
extra_join_parts: list[str] | None = None,
join_parts: list[str],
favorite: bool | None,
search: str | None,
- provider: str | None,
+ provider: str | list[str] | None,
limit: int,
) -> None:
"""Build a fast random subquery with all filters applied."""
join_parts: list[str],
favorite: bool | None,
search: str | None,
- provider: str | None,
+ provider: str | list[str] | None,
) -> None:
- """Apply search, favorite, and provider filters."""
+ """Apply search, favorite, and provider filters.
+
+ If the current user has a provider_filter set, it will be applied automatically.
+ If an explicit provider filter is provided, it will be validated against the user's
+ allowed providers.
+ """
# handle search
if search:
query_parts.append(f"{self.db_table}.search_name LIKE :search")
query_parts.append(f"{self.db_table}.favorite = :favorite")
query_params["favorite"] = favorite
- # handle provider filter
- if provider:
+ # Apply user provider filter
+ user = get_current_user()
+ user_provider_filter = user.provider_filter if user and user.provider_filter else None
+
+ # Determine final provider filter
+ final_provider_filter: list[str] | None = None
+
+ if user_provider_filter:
+ # User has a provider filter set
+ if provider:
+ # Explicit provider filter provided - validate against user's allowed providers
+ requested_providers = [provider] if isinstance(provider, str) else provider
+ # Only include providers that are in both the user's filter and the requested list
+ final_provider_filter = [
+ p for p in requested_providers if p in user_provider_filter
+ ]
+ if not final_provider_filter:
+ # No overlap - user requested providers they don't have access to
+ # Return empty results by adding impossible condition
+ query_parts.append("1 = 0")
+ return
+ else:
+ # No explicit filter - use user's provider filter
+ final_provider_filter = user_provider_filter
+ elif provider:
+ # No user filter, but explicit provider filter provided
+ final_provider_filter = [provider] if isinstance(provider, str) else provider
+
+ # Apply the final provider filter
+ if final_provider_filter:
+ provider_conditions = []
+ for prov in final_provider_filter:
+ provider_conditions.append(
+ f"provider_mappings.provider_instance = '{prov}' "
+ f"OR provider_mappings.provider_domain = '{prov}'"
+ )
join_parts.append(
f"JOIN provider_mappings ON provider_mappings.item_id = {self.db_table}.item_id "
f"AND provider_mappings.media_type = '{self.media_type.value}' "
"AND provider_mappings.in_library = 1 "
- f"AND (provider_mappings.provider_instance = '{provider}' "
- f"OR provider_mappings.provider_domain = '{provider}')"
+ f"AND ({' OR '.join(provider_conditions)})"
)
def _build_final_query(
limit: int = 500,
offset: int = 0,
order_by: str = "sort_name",
- provider: str | None = None,
+ provider: str | list[str] | None = None,
extra_query: str | None = None,
extra_query_params: dict[str, Any] | None = None,
) -> list[Podcast]:
- """Get in-database podcasts."""
+ """Get in-database podcasts.
+
+ :param favorite: Filter by favorite status.
+ :param search: Filter by search query.
+ :param limit: Maximum number of items to return.
+ :param offset: Number of items to skip.
+ :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
+ :param provider: Filter by provider instance ID or domain (single string or list).
+ :param extra_query: Additional SQL query string.
+ :param extra_query_params: Additional query parameters.
+ """
extra_query_params = extra_query_params or {}
extra_query_parts: list[str] = [extra_query] if extra_query else []
result = await self._get_library_items_by_query(
limit: int = 500,
offset: int = 0,
order_by: str = "sort_name",
- provider: str | None = None,
+ provider: str | list[str] | None = None,
extra_query: str | None = None,
extra_query_params: dict[str, Any] | None = None,
) -> list[Track]:
- """Get in-database tracks."""
+ """Get in-database tracks.
+
+ :param favorite: Filter by favorite status.
+ :param search: Filter by search query.
+ :param limit: Maximum number of items to return.
+ :param offset: Number of items to skip.
+ :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
+ :param provider: Filter by provider instance ID or domain (single string or list).
+ :param extra_query: Additional SQL query string.
+ :param extra_query_params: Additional query parameters.
+ """
extra_query_params = extra_query_params or {}
extra_query_parts: list[str] = [extra_query] if extra_query else []
extra_join_parts: list[str] = []
x
for x in self._providers.values()
if (provider_type is None or provider_type == x.type)
- and (not user_provider_filter or x.instance_id in user_provider_filter)
+ # handle optional user (music) provider filter
+ and (
+ not user_provider_filter
+ or x.instance_id in user_provider_filter
+ or x.type != ProviderType.MUSIC
+ )
]
@api_command("logging/get")
projects / music-assistant-server.git / commitdiff