import asyncio
import os
-import webbrowser
from collections.abc import AsyncGenerator
from logging import getLevelName
from typing import TYPE_CHECKING, cast
ConfigValueType,
ProviderConfig,
)
-from music_assistant_models.enums import ConfigEntryType, MediaType, ProviderFeature
+from music_assistant_models.enums import ConfigEntryType, EventType, MediaType, ProviderFeature
from music_assistant_models.errors import LoginFailed
from music_assistant.models.music_provider import MusicProvider
values[CONF_CODE_VERIFIER] = code_verifier
values[CONF_SERIAL] = serial
values[CONF_LOGIN_URL] = login_url
- webbrowser.open_new_tab(login_url)
+ session_id = str(values["session_id"])
+ mass.signal_event(EventType.AUTH_SESSION, session_id, login_url)
+ await asyncio.sleep(15)
if action == CONF_ACTION_VERIFY:
code_verifier = str(values.get(CONF_CODE_VERIFIER))
while True:
library = await self._call_api(
"library",
+ use_cache=False,
response_groups=",".join(response_groups),
page=page,
num_results=page_size,
"""Report last position."""
async def _call_api(self, path: str, **kwargs: Any) -> Any:
+ response = None
+ use_cache = False
params_str = json.dumps(kwargs, sort_keys=True)
params_hash = hashlib.md5(params_str.encode()).hexdigest()
cache_key_with_params = f"{path}:{params_hash}"
-
- response = await self.mass.cache.get(
- key=cache_key_with_params, base_key=CACHE_DOMAIN, category=CACHE_CATEGORY_API
- )
+ if use_cache:
+ response = await self.mass.cache.get(
+ key=cache_key_with_params, base_key=CACHE_DOMAIN, category=CACHE_CATEGORY_API
+ )
if not response:
response = await self.client.get(path, **kwargs)
await self.mass.cache.set(