ProviderFeature,
StreamType,
)
-from music_assistant_models.errors import MediaNotFoundError
+from music_assistant_models.errors import LoginFailed, MediaNotFoundError
from music_assistant_models.media_items import (
Audiobook,
AudioFormat,
async def handle_async_init(self) -> None:
"""Pass config values to client and initialize."""
self._client = ABSClient()
- await self._client.init(
- session=self.mass.http_session,
- base_url=str(self.config.get_value(CONF_URL)),
- username=str(self.config.get_value(CONF_USERNAME)),
- password=str(self.config.get_value(CONF_PASSWORD)),
- check_ssl=bool(self.config.get_value(CONF_VERIFY_SSL)),
- )
+ base_url = str(self.config.get_value(CONF_URL))
+ try:
+ await self._client.init(
+ session=self.mass.http_session,
+ base_url=base_url,
+ username=str(self.config.get_value(CONF_USERNAME)),
+ password=str(self.config.get_value(CONF_PASSWORD)),
+ check_ssl=bool(self.config.get_value(CONF_VERIFY_SSL)),
+ )
+ except RuntimeError:
+ # login details were not correct
+ raise LoginFailed(f"Login to abs instance at {base_url} failed.")
await self._client.sync()
async def unload(self, is_removed: bool = False) -> None:
"""Logout from ABS."""
await self._post("logout", add_api_endpoint=False)
- async def get_user(self, id_: str) -> ABSUser:
+ async def get_authenticated_user(self) -> ABSUser:
"""Get an ABS user."""
- data = await self._get(f"users/{id_}")
+ data = await self._get("me")
return ABSUser.from_json(data)
async def sync(self) -> None:
self.audiobook_libraries.append(library)
elif media_type == "podcast":
self.podcast_libraries.append(library)
- self.user = await self.get_user(self.user.id_)
+ self.user = await self.get_authenticated_user()
async def get_all_podcasts(self) -> AsyncGenerator[ABSPodcast]:
"""Get all available podcasts."""