Merge pull request mopidy#2119 from mopidy/typing-m3u

Type mopidy.m3u

mopidy/m3u/__init__.py

@@ -3,6 +3,7 @@
 
 import mopidy
 from mopidy import config, ext
+from mopidy.config import ConfigSchema
 
 logger = logging.getLogger(__name__)
 
@@ -12,18 +13,18 @@ class Extension(ext.Extension):
     ext_name = "m3u"
     version = mopidy.__version__
 
-    def get_default_config(self):
+    def get_default_config(self) -> str:
         return config.read(Path(__file__).parent / "ext.conf")
 
-    def get_config_schema(self):
+    def get_config_schema(self) -> ConfigSchema:
         schema = super().get_config_schema()
         schema["base_dir"] = config.Path(optional=True)
         schema["default_encoding"] = config.String()
         schema["default_extension"] = config.String(choices=[".m3u", ".m3u8"])
         schema["playlists_dir"] = config.Path(optional=True)
         return schema
 
-    def setup(self, registry):
+    def setup(self, registry: ext.Registry) -> None:
         from .backend import M3UBackend
 
         registry.add("backend", M3UBackend)

mopidy/m3u/backend.py

@@ -1,17 +1,26 @@
-# ruff: noqa: ARG002
+from __future__ import annotations
 
-from typing import ClassVar
+from typing import TYPE_CHECKING, ClassVar
 
 import pykka
 
 from mopidy import backend
+from mopidy.types import UriScheme
 
 from . import playlists
 
+if TYPE_CHECKING:
+    from mopidy.audio import AudioProxy
+    from mopidy.ext import Config
+
 
 class M3UBackend(pykka.ThreadingActor, backend.Backend):
-    uri_schemes: ClassVar[list[str]] = ["m3u"]
+    uri_schemes: ClassVar[list[UriScheme]] = [UriScheme("m3u")]
 
-    def __init__(self, config, audio):
+    def __init__(
+        self,
+        config: Config,
+        audio: AudioProxy,  # noqa: ARG002
+    ) -> None:
         super().__init__()
         self.playlists = playlists.M3UPlaylistsProvider(self, config)

mopidy/m3u/playlists.py

@@ -5,23 +5,28 @@
 import logging
 import operator
 import os
-import pathlib
 import tempfile
-from typing import TYPE_CHECKING
+from collections.abc import Generator
+from pathlib import Path
+from typing import IO, TYPE_CHECKING, Any, Optional, Union, cast
 
 from mopidy import backend
 from mopidy.exceptions import BackendError
 from mopidy.internal import path
+from mopidy.m3u.types import M3UConfig
 
 from . import Extension, translator
 
 if TYPE_CHECKING:
-    from mopidy.models import Playlist
+    from mopidy.backend import Backend
+    from mopidy.ext import Config
+    from mopidy.models import Playlist, Ref
+    from mopidy.types import Uri
 
 logger = logging.getLogger(__name__)
 
 
-def log_environment_error(message, error):
+def log_environment_error(message: str, error: EnvironmentError) -> None:
     if isinstance(error.strerror, bytes):
         strerror = error.strerror.decode(locale.getpreferredencoding())
     else:
@@ -30,9 +35,14 @@ def log_environment_error(message, error):
 
 
 @contextlib.contextmanager
-def replace(path, mode="w+b", encoding=None, errors=None):
+def replace(
+    path: Path,
+    mode: str = "w+b",
+    encoding: Optional[str] = None,
+    errors: Optional[str] = None,
+) -> Generator[IO[Any], None, None]:
     (fd, tempname) = tempfile.mkstemp(dir=str(path.parent))
-    tempname = pathlib.Path(tempname)
+    tempname = Path(tempname)
     try:
         fp = open(fd, mode, encoding=encoding, errors=errors)  # noqa: PTH123, SIM115
     except Exception:
@@ -52,22 +62,25 @@ def replace(path, mode="w+b", encoding=None, errors=None):
 
 
 class M3UPlaylistsProvider(backend.PlaylistsProvider):
-    def __init__(self, backend, config):
+    def __init__(self, backend: Backend, config: Config) -> None:
         super().__init__(backend)
 
-        ext_config = config[Extension.ext_name]
-        if ext_config["playlists_dir"] is None:
-            self._playlists_dir = Extension.get_data_dir(config)
-        else:
-            self._playlists_dir = path.expand_path(ext_config["playlists_dir"])
-        if ext_config["base_dir"] is None:
-            self._base_dir = self._playlists_dir
-        else:
-            self._base_dir = path.expand_path(ext_config["base_dir"])
+        ext_config = cast(M3UConfig, config[Extension.ext_name])
+
+        self._playlists_dir = (
+            path.expand_path(ext_config["playlists_dir"])
+            if ext_config["playlists_dir"]
+            else Extension.get_data_dir(config)
+        )
+        self._base_dir = (
+            path.expand_path(ext_config["base_dir"])
+            if ext_config["base_dir"]
+            else self._playlists_dir
+        )
         self._default_encoding = ext_config["default_encoding"]
         self._default_extension = ext_config["default_extension"]
 
-    def as_list(self):
+    def as_list(self) -> list[Ref]:
         result = []
         for entry in self._playlists_dir.iterdir():
             if entry.suffix not in [".m3u", ".m3u8"]:
@@ -79,7 +92,7 @@ def as_list(self):
         result.sort(key=operator.attrgetter("name"))
         return result
 
-    def create(self, name):
+    def create(self, name: str) -> Optional[Playlist]:
         path = translator.path_from_name(name.strip(), self._default_extension)
         try:
             with self._open(path, "w"):
@@ -90,7 +103,7 @@ def create(self, name):
         else:
             return translator.playlist(path, [], mtime)
 
-    def delete(self, uri):
+    def delete(self, uri: Uri) -> bool:
         path = translator.uri_to_path(uri)
         if not self._is_in_basedir(path):
             logger.debug("Ignoring path outside playlist dir: %s", uri)
@@ -103,7 +116,7 @@ def delete(self, uri):
         else:
             return True
 
-    def get_items(self, uri):
+    def get_items(self, uri: Uri) -> Optional[list[Ref]]:
         path = translator.uri_to_path(uri)
         if not self._is_in_basedir(path):
             logger.debug("Ignoring path outside playlist dir: %s", uri)
@@ -116,7 +129,7 @@ def get_items(self, uri):
         else:
             return items
 
-    def lookup(self, uri):
+    def lookup(self, uri: Uri) -> Optional[Playlist]:
         path = translator.uri_to_path(uri)
         if not self._is_in_basedir(path):
             logger.debug("Ignoring path outside playlist dir: %s", uri)
@@ -130,10 +143,10 @@ def lookup(self, uri):
         else:
             return translator.playlist(path, items, mtime)
 
-    def refresh(self):
+    def refresh(self) -> None:
         pass  # nothing to do
 
-    def save(self, playlist: Playlist) -> Playlist | None:
+    def save(self, playlist: Playlist) -> Optional[Playlist]:
         path = translator.uri_to_path(playlist.uri)
         if not self._is_in_basedir(path):
             logger.debug("Ignoring path outside playlist dir: %s", playlist.uri)
@@ -153,16 +166,18 @@ def save(self, playlist: Playlist) -> Playlist | None:
         else:
             return translator.playlist(path, playlist.tracks, mtime)
 
-    def _abspath(self, path):
+    def _abspath(self, path: Path) -> Path:
         if path.is_absolute():
             return path
         return self._playlists_dir / path
 
-    def _is_in_basedir(self, local_path):
+    def _is_in_basedir(self, local_path: Path) -> bool:
         local_path = self._abspath(local_path)
         return path.is_path_inside_base_dir(local_path, self._playlists_dir)
 
-    def _open(self, path, mode="r"):
+    def _open(
+        self, path: Path, mode: str = "r"
+    ) -> Union[contextlib._GeneratorContextManager[IO[Any]], IO[Any]]:
         encoding = "utf-8" if path.suffix == ".m3u8" else self._default_encoding
         if not path.is_absolute():
             path = self._abspath(path)

mopidy/m3u/translator.py

@@ -1,45 +1,60 @@
+from __future__ import annotations
+
 import os
-import pathlib
 import urllib.parse
+from collections.abc import Iterable
+from pathlib import Path
+from typing import IO, Optional, Union
 
-from mopidy import models
 from mopidy.internal import path
+from mopidy.models import Playlist, Ref, Track
+from mopidy.types import Uri
 
 from . import Extension
 
 
-def path_to_uri(path, scheme=Extension.ext_name):
+def path_to_uri(
+    path: Path,
+    scheme: str = Extension.ext_name,
+) -> Uri:
     """Convert file path to URI."""
     bytes_path = os.path.normpath(bytes(path))
     uripath = urllib.parse.quote_from_bytes(bytes_path)
-    return urllib.parse.urlunsplit((scheme, None, uripath, None, None))
+    return Uri(urllib.parse.urlunsplit((scheme, None, uripath, None, None)))
 
 
-def uri_to_path(uri):
+def uri_to_path(uri: Uri) -> Path:
     """Convert URI to file path."""
     return path.uri_to_path(uri)
 
 
-def name_from_path(path):
+def name_from_path(path: Path) -> Optional[str]:
     """Extract name from file path."""
-    name = bytes(pathlib.Path(path.stem))
+    name = bytes(Path(path.stem))
     try:
         return name.decode(errors="replace")
     except UnicodeError:
         return None
 
 
-def path_from_name(name, ext=None, sep="|"):
+def path_from_name(
+    name: str,
+    ext: Optional[str] = None,
+    sep: str = "|",
+) -> Path:
     """Convert name with optional extension to file path."""
     name = name.replace(os.sep, sep) + ext if ext else name.replace(os.sep, sep)
-    return pathlib.Path(name)
+    return Path(name)
 
 
-def path_to_ref(path):
-    return models.Ref.playlist(uri=path_to_uri(path), name=name_from_path(path))
+def path_to_ref(path: Path) -> Ref:
+    return Ref.playlist(uri=path_to_uri(path), name=name_from_path(path))
 
 
-def load_items(fp, basedir):
+def load_items(
+    fp: IO[str],
+    basedir: Path,
+) -> list[Ref]:
     refs = []
     name = None
     for line in filter(None, (line.strip() for line in fp)):
@@ -55,12 +70,15 @@ def load_items(fp, basedir):
         else:
             # TODO: ensure this is urlencoded
             uri = line  # do *not* extract name from (stream?) URI path
-        refs.append(models.Ref.track(uri=uri, name=name))
+        refs.append(Ref.track(uri=uri, name=name))
         name = None
     return refs
 
 
-def dump_items(items, fp):
+def dump_items(
+    items: Iterable[Union[Ref, Track]],
+    fp: IO[str],
+) -> None:
     if any(item.name for item in items):
         print("#EXTM3U", file=fp)
     for item in items:
@@ -73,12 +91,16 @@ def dump_items(items, fp):
             print(item.uri, file=fp)
 
 
-def playlist(path, items=None, mtime=None):
+def playlist(
+    path: Path,
+    items: Optional[Iterable[Union[Ref, Track]]] = None,
+    mtime: Optional[float] = None,
+) -> Playlist:
     if items is None:
         items = []
-    return models.Playlist(
+    return Playlist(
         uri=path_to_uri(path),
         name=name_from_path(path),
-        tracks=[models.Track(uri=item.uri, name=item.name) for item in items],
+        tracks=[Track(uri=item.uri, name=item.name) for item in items],
         last_modified=(int(mtime * 1000) if mtime else None),
     )

mopidy/m3u/types.py

@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Literal, Optional, TypedDict
+
+
+class M3UConfig(TypedDict):
+    base_dir: Optional[Path]
+    default_encoding: str
+    default_extension: Literal[".m3u", ".m3u8"]
+    playlists_dir: Optional[Path]