Add typing for filesystem_local (#1461)

music_assistant/common/models/media_items.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Iterable
+from collections.abc import Iterable, Sequence
 from dataclasses import dataclass, field, fields
 from typing import TYPE_CHECKING, Any, TypeGuard, TypeVar, cast
 
@@ -526,11 +526,11 @@ def __post_init__(self) -> None:
 class SearchResults(DataClassDictMixin):
     """Model for results from a search query."""
 
-    artists: list[Artist | ItemMapping] = field(default_factory=list)
-    albums: list[Album | ItemMapping] = field(default_factory=list)
-    tracks: list[Track | ItemMapping] = field(default_factory=list)
-    playlists: list[Playlist | ItemMapping] = field(default_factory=list)
-    radio: list[Radio | ItemMapping] = field(default_factory=list)
+    artists: Sequence[Artist | ItemMapping] = field(default_factory=list)
+    albums: Sequence[Album | ItemMapping] = field(default_factory=list)
+    tracks: Sequence[Track | ItemMapping] = field(default_factory=list)
+    playlists: Sequence[Playlist | ItemMapping] = field(default_factory=list)
+    radio: Sequence[Radio | ItemMapping] = field(default_factory=list)
 
 
 def media_from_dict(media_item: dict[str, Any]) -> MediaItemType | ItemMapping:

music_assistant/server/controllers/media/base.py

@@ -91,7 +91,7 @@ def __init__(self, mass: MusicAssistant) -> None:
         self._db_add_lock = asyncio.Lock()
 
     async def add_item_to_library(
-        self, item: Track, metadata_lookup: bool = True, overwrite_existing: bool = False
+        self, item: ItemCls, metadata_lookup: bool = True, overwrite_existing: bool = False
     ) -> ItemCls:
         """Add item to library and return the new (or updated) database item."""
         new_item = False
@@ -758,7 +758,7 @@ async def _get_library_items_by_query(
         provider: str | None = None,
         extra_query: str | None = None,
         extra_query_params: dict[str, Any] | None = None,
-    ) -> list[ItemCls] | int:
+    ) -> list[ItemCls]:
         """Fetch MediaItem records from database given a custom (WHERE) clause."""
         sql_query = self.base_query
         query_params = extra_query_params or {}

music_assistant/server/models/music_provider.py

@@ -12,6 +12,7 @@
     Album,
     Artist,
     BrowseFolder,
+    ItemMapping,
     MediaItemType,
     Playlist,
     Radio,
@@ -257,6 +258,8 @@ async def get_audio_stream(  # type: ignore[return]
 
         Will only be called when the stream_type is set to CUSTOM.
         """
+        if False:
+            yield
         raise NotImplementedError
 
     async def on_streamed(self, streamdetails: StreamDetails, seconds_streamed: int) -> None:
@@ -283,7 +286,9 @@ async def get_item(self, media_type: MediaType, prov_item_id: str) -> MediaItemT
             return await self.get_radio(prov_item_id)
         return await self.get_track(prov_item_id)
 
-    async def browse(self, path: str, offset: int, limit: int) -> Sequence[MediaItemType]:
+    async def browse(
+        self, path: str, offset: int, limit: int
+    ) -> Sequence[MediaItemType | ItemMapping]:
         """Browse this provider's items.
 
         :param path: The path to browse, (e.g. provider_id://artists).

music_assistant/server/providers/filesystem_local/__init__.py

@@ -51,7 +51,7 @@ async def setup(
         msg = f"Music Directory {conf_path} does not exist"
         raise SetupFailedError(msg)
     prov = LocalFileSystemProvider(mass, manifest, config)
-    prov.base_path = config.get_value(CONF_PATH)
+    prov.base_path = str(config.get_value(CONF_PATH))
     await prov.check_write_access()
     mass.call_later(30, prov.migrate_playlists)
     return prov
@@ -77,14 +77,14 @@ async def get_config_entries(
     )
 
 
-def sorted_scandir(base_path: str, sub_path: str) -> list[os.DirEntry]:
+def sorted_scandir(base_path: str, sub_path: str) -> list[FileSystemItem]:
     """Implement os.scandir that returns (naturally) sorted entries."""
 
-    def nat_key(name: str) -> tuple[int, str]:
+    def nat_key(name: str) -> tuple[int | str, ...]:
         """Sort key for natural sorting."""
         return tuple(int(s) if s.isdigit() else s for s in re.split(r"(\d+)", name))
 
-    def create_item(entry: os.DirEntry):
+    def create_item(entry: os.DirEntry) -> FileSystemItem:
         """Create FileSystemItem from os.DirEntry."""
         absolute_path = get_absolute_path(base_path, entry.path)
         stat = entry.stat(follow_symlinks=False)
@@ -168,7 +168,7 @@ async def resolve(
         """
         absolute_path = get_absolute_path(self.base_path, file_path)
 
-        def _create_item():
+        def _create_item() -> FileSystemItem:
             stat = os.stat(absolute_path, follow_symlinks=False)
             return FileSystemItem(
                 filename=os.path.basename(file_path),
@@ -190,7 +190,7 @@ async def exists(self, file_path: str) -> bool:
         if not file_path:
             return False  # guard
         abs_path = get_absolute_path(self.base_path, file_path)
-        return await exists(abs_path)
+        return bool(await exists(abs_path))
 
     async def read_file_content(self, file_path: str, seek: int = 0) -> AsyncGenerator[bytes, None]:
         """Yield (binary) contents of file in chunks of bytes."""
@@ -229,11 +229,11 @@ async def migrate_playlists(self) -> None:
                 continue
             if item.ext != "m3u":
                 continue
-            playlist_data = b""
+            playlist_bytes = b""
             async for chunk in self.read_file_content(item.absolute_path):
-                playlist_data += chunk
-            encoding_details = await asyncio.to_thread(cchardet.detect, playlist_data)
-            playlist_data = playlist_data.decode(encoding_details["encoding"] or "utf-8")
+                playlist_bytes += chunk
+            encoding_details = await asyncio.to_thread(cchardet.detect, playlist_bytes)
+            playlist_data = playlist_bytes.decode(encoding_details["encoding"] or "utf-8")
             # a (legacy) playlist file created by MA does not have EXTINFO tags and has uri's
             if "EXTINF" in playlist_data or "://" not in playlist_data:
                 continue