if cache := await self.mass.cache.get(cache_key, checksum=cache_checksum):
return [Track.from_dict(x) for x in cache]
# no items in cache - get listing from provider
- items = await prov.get_artist_toptracks(item_id)
+ if MusicProviderFeature.ARTIST_TOPTRACKS in prov.supported_features:
+ items = await prov.get_artist_toptracks(item_id)
+ else:
+ # fallback implementation using the db
+ if db_artist := await self.mass.music.artists.get_db_item_by_prov_id(
+ item_id, provider=provider, provider_id=provider_id
+ ):
+ prov_id = provider_id or provider.value
+ # TODO: adjust to json query instead of text search?
+ query = f"SELECT * FROM tracks WHERE artists LIKE '%\"{db_artist.item_id}\"%'"
+ query += f" AND provider_ids LIKE '%\"{prov_id}\"%'"
+ items = await self.mass.music.tracks.get_db_items_by_query(query)
# store (serializable items) in cache
self.mass.create_task(
self.mass.cache.set(
if cache := await self.mass.cache.get(cache_key, checksum=cache_checksum):
return [Album.from_dict(x) for x in cache]
# no items in cache - get listing from provider
- items = await prov.get_artist_albums(item_id)
+ if MusicProviderFeature.ARTIST_ALBUMS in prov.supported_features:
+ items = await prov.get_artist_albums(item_id)
+ else:
+ # fallback implementation using the db
+ if db_artist := await self.mass.music.artists.get_db_item_by_prov_id(
+ item_id, provider=provider, provider_id=provider_id
+ ):
+ prov_id = provider_id or provider.value
+ # TODO: adjust to json query instead of text search?
+ query = f"SELECT * FROM albums WHERE artists LIKE '%\"{db_artist.item_id}\"%'"
+ query += f" AND provider_ids LIKE '%\"{prov_id}\"%'"
+ items = await self.mass.music.albums.get_db_items_by_query(query)
# store (serializable items) in cache
self.mass.create_task(
self.mass.cache.set(
if MusicProviderFeature.PLAYLIST_TRACKS_EDIT in self.supported_features:
raise NotImplementedError
+ async def create_playlist(
+ self, name: str, initial_items: Optional[List[Track]] = None
+ ) -> Playlist:
+ """Create a new playlist on provider with given name."""
+ raise NotImplementedError
+
async def get_stream_details(self, item_id: str) -> StreamDetails | None:
"""Get streamdetails for a track/radio."""
raise NotImplementedError
MusicProviderFeature.LIBRARY_ALBUMS,
MusicProviderFeature.LIBRARY_TRACKS,
MusicProviderFeature.LIBRARY_PLAYLISTS,
- MusicProviderFeature.LIBRARY_RADIOS,
- MusicProviderFeature.LIBRARY_ARTISTS_EDIT,
- MusicProviderFeature.LIBRARY_ALBUMS_EDIT,
- MusicProviderFeature.LIBRARY_PLAYLISTS_EDIT,
- MusicProviderFeature.LIBRARY_RADIOS_EDIT,
- MusicProviderFeature.LIBRARY_TRACKS_EDIT,
MusicProviderFeature.PLAYLIST_TRACKS_EDIT,
+ MusicProviderFeature.PLAYLIST_CREATE,
MusicProviderFeature.BROWSE,
MusicProviderFeature.SEARCH,
- MusicProviderFeature.ARTIST_ALBUMS,
- MusicProviderFeature.ARTIST_TOPTRACKS,
)
async def setup(self) -> bool:
)
return None
- async def get_artist_albums(self, prov_artist_id: str) -> List[Album]:
- """Get a list of albums for the given artist."""
- # filesystem items are always stored in db so we can query the database
- db_artist = await self.mass.music.artists.get_db_item_by_prov_id(
- prov_artist_id, provider_id=self.id
- )
- if db_artist is None:
- raise MediaNotFoundError(f"Artist not found: {prov_artist_id}")
- # TODO: adjust to json query instead of text search
- query = f"SELECT * FROM albums WHERE artists LIKE '%\"{db_artist.item_id}\"%'"
- query += f" AND provider_ids LIKE '%\"{self.type.value}\"%'"
- return await self.mass.music.albums.get_db_items_by_query(query)
-
- async def get_artist_toptracks(self, prov_artist_id: str) -> List[Track]:
- """Get a list of all tracks as we have no clue about preference."""
- # filesystem items are always stored in db so we can query the database
- db_artist = await self.mass.music.artists.get_db_item_by_prov_id(
- prov_artist_id, provider_id=self.id
- )
- if db_artist is None:
- raise MediaNotFoundError(f"Artist not found: {prov_artist_id}")
- # TODO: adjust to json query instead of text search
- query = f"SELECT * FROM tracks WHERE artists LIKE '%\"{db_artist.item_id}\"%'"
- query += f" AND provider_ids LIKE '%\"{self.type.value}\"%'"
- return await self.mass.music.tracks.get_db_items_by_query(query)
-
- async def library_add(self, *args, **kwargs) -> bool:
- """Add item to provider's library. Return true on succes."""
- # already handled by database
-
- async def library_remove(self, *args, **kwargs) -> bool:
- """Remove item from provider's library. Return true on succes."""
- # already handled by database
- # TODO: do we want to process/offer deletions here ?
-
async def add_playlist_tracks(
self, prov_playlist_id: str, prov_track_ids: List[str]
) -> None:
for uri in cur_lines:
await _file.write(f"{uri}\n")
+ async def create_playlist(
+ self, name: str, initial_items: Optional[List[Track]] = None
+ ) -> Playlist:
+ """Create a new playlist on provider with given name."""
+ # creating a new playlist on the filesystem is as easy
+ # as creating a new (empty) file with the m3u extension...
+ async with self.open_file(name, "w") as _file:
+ for item in initial_items or []:
+ await _file.write(item.uri + "\n")
+ await _file.write("\n")
+
async def get_stream_details(self, item_id: str) -> StreamDetails:
"""Return the content details for the given track when it will be streamed."""
itempath = await self._get_filepath(MediaType.TRACK, item_id)
MusicProviderFeature.LIBRARY_ALBUMS,
MusicProviderFeature.LIBRARY_TRACKS,
MusicProviderFeature.LIBRARY_PLAYLISTS,
- MusicProviderFeature.LIBRARY_RADIOS,
MusicProviderFeature.LIBRARY_ARTISTS_EDIT,
MusicProviderFeature.LIBRARY_ALBUMS_EDIT,
MusicProviderFeature.LIBRARY_PLAYLISTS_EDIT,
- MusicProviderFeature.LIBRARY_RADIOS_EDIT,
MusicProviderFeature.LIBRARY_TRACKS_EDIT,
MusicProviderFeature.PLAYLIST_TRACKS_EDIT,
MusicProviderFeature.BROWSE,
MusicProviderFeature.LIBRARY_ALBUMS,
MusicProviderFeature.LIBRARY_TRACKS,
MusicProviderFeature.LIBRARY_PLAYLISTS,
- MusicProviderFeature.LIBRARY_RADIOS,
MusicProviderFeature.LIBRARY_ARTISTS_EDIT,
MusicProviderFeature.LIBRARY_ALBUMS_EDIT,
MusicProviderFeature.LIBRARY_PLAYLISTS_EDIT,
- MusicProviderFeature.LIBRARY_RADIOS_EDIT,
MusicProviderFeature.LIBRARY_TRACKS_EDIT,
MusicProviderFeature.PLAYLIST_TRACKS_EDIT,
MusicProviderFeature.BROWSE,
ImageType,
MediaQuality,
MediaType,
- MusicProviderFeature,
ProviderType,
)
from music_assistant.models.media_items import (
_attr_available: bool = True
_full_url = {}
- @property
- def supported_features(self) -> Tuple[MusicProviderFeature]:
- """Return the features supported by this MusicProvider."""
- # return empty tuple because we do not really support features directly here
- return tuple()
-
async def setup(self) -> bool:
"""
Handle async initialization of the provider.
projects / music-assistant-server.git / commitdiff