if queue.enqueued_media_items:
# we need to restore the MediaItem objects for the enqueued media items
# Items from cache may be dicts that need deserialization
-
restored_enqueued_items: list[MediaItemType] = []
- cached_items: list[Any] = cast("list[Any]", queue.enqueued_media_items)
+ cached_items: list[dict[str, Any] | MediaItemType] = cast(
+ "list[dict[str, Any] | MediaItemType]", queue.enqueued_media_items
+ )
for item in cached_items:
if isinstance(item, dict):
restored_item = media_from_dict(item)
from typing import Any, TypeVar, Union, get_args, get_origin, get_type_hints
from mashumaro.exceptions import MissingField
+from music_assistant_models.media_items.media_item import MediaItem
from music_assistant.helpers.util import try_parse_bool
return value
if isinstance(value, dict) and hasattr(value_type, "from_dict"):
+ # Only validate media_type for actual MediaItem subclasses, not for other classes
+ # like StreamDetails that have a media_type field for a different purpose
if (
"media_type" in value
and value_type.__name__ != "ItemMapping"
+ and issubclass(value_type, MediaItem)
and value["media_type"] != value_type.media_type
):
msg = "Invalid MediaType"
raise MediaNotFoundError(f"Episode {prov_episode_id} not found")
- @use_cache(86400)
async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
"""
Get stream details for a podcast episode.
if mass_episode := self._parse_episode(episode, idx):
yield mass_episode
- @use_cache(3600) # Cache for 1 hour
async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
"""Get streamdetails for a track/radio."""
for episode in self.parsed_podcast["episodes"]:
return tracks
- @use_cache(3600 * 3) # Cache for 3 hours
async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
"""Return the content details for the given track when it will be streamed."""
url: str = await self._soundcloud.get_stream_url(track_id=item_id, presets=["mp3"])