refactor: add type hints

m3u8/__init__.py

@@ -3,9 +3,11 @@
 # license that can be found in the LICENSE file.
 
 import os
+from collections.abc import Callable, Mapping
+from typing import Any
 from urllib.parse import urljoin, urlsplit
 
-from m3u8.httpclient import DefaultHTTPClient
+from m3u8.httpclient import DefaultHTTPClient, _HTTPClientProtocol
 from m3u8.model import (
     M3U8,
     ContentSteering,
@@ -64,7 +66,14 @@
 )
 
 
-def loads(content, uri=None, custom_tags_parser=None):
+type _CustomTagsParser = Callable[[str, int, dict[str, Any], dict[str, Any]], object]
+
+
+def loads(
+    content: str,
+    uri: str | None = None,
+    custom_tags_parser: _CustomTagsParser | None = None,
+) -> M3U8:
     """
     Given a string with a m3u8 content, returns a M3U8 object.
     Optionally parses a uri to set a correct base_uri on the M3U8 object.
@@ -79,26 +88,30 @@ def loads(content, uri=None, custom_tags_parser=None):
 
 
 def load(
-    uri,
-    timeout=None,
-    headers={},
-    custom_tags_parser=None,
-    http_client=DefaultHTTPClient(),
-    verify_ssl=True,
-):
+    uri: str,
+    timeout: float | None = None,
+    headers: Mapping[str, Any] | None = None,
+    custom_tags_parser: _CustomTagsParser | None = None,
+    http_client: _HTTPClientProtocol = DefaultHTTPClient(),
+    verify_ssl: bool = True,
+) -> M3U8:
     """
     Retrieves the content from a given URI and returns a M3U8 object.
     Raises ValueError if invalid content or IOError if request fails.
     """
     base_uri_parts = urlsplit(uri)
     if base_uri_parts.scheme and base_uri_parts.netloc:
-        content, base_uri = http_client.download(uri, timeout, headers, verify_ssl)
+        content, base_uri = http_client.download(
+            uri, timeout, headers or {}, verify_ssl
+        )
         return M3U8(content, base_uri=base_uri, custom_tags_parser=custom_tags_parser)
-    else:
-        return _load_from_file(uri, custom_tags_parser)
+
+    return _load_from_file(uri, custom_tags_parser)
 
 
-def _load_from_file(uri, custom_tags_parser=None):
+def _load_from_file(
+    uri: os.PathLike[str] | str, custom_tags_parser: _CustomTagsParser | None = None
+) -> M3U8:
     with open(uri, encoding="utf8") as fileobj:
         raw_content = fileobj.read().strip()
     base_uri = os.path.dirname(uri)

m3u8/httpclient.py

@@ -1,18 +1,38 @@
 import gzip
 import ssl
 import urllib.request
+from collections.abc import Mapping
+from typing import Any, Protocol
 from urllib.parse import urljoin
 
 
-class DefaultHTTPClient:
-    def __init__(self, proxies=None):
-        self.proxies = proxies
+class _HTTPClientProtocol(Protocol):  # noqa: Y046
+    def download(
+        self,
+        uri: str,
+        timeout: float | None = None,
+        headers: Mapping[str, Any] | None = None,
+        verify_ssl: bool = True,
+    ) -> tuple[str, str]: ...
+
+
+class DefaultHTTPClient(_HTTPClientProtocol):
+    def __init__(self, proxies: dict[str, str] | None = None) -> None:
+        self.proxies: dict[str, str] | None = proxies
+
+    def download(
+        self,
+        uri: str,
+        timeout: float | None = None,
+        headers: Mapping[str, Any] | None = None,
+        verify_ssl: bool = True,
+    ) -> tuple[str, str]:
 
-    def download(self, uri, timeout=None, headers={}, verify_ssl=True):
         proxy_handler = urllib.request.ProxyHandler(self.proxies)
         https_handler = HTTPSHandler(verify_ssl=verify_ssl)
         opener = urllib.request.build_opener(proxy_handler, https_handler)
-        opener.addheaders = headers.items()
+        if headers:
+            opener.addheaders = list(headers.items())
         resource = opener.open(uri, timeout=timeout)
         base_uri = urljoin(resource.geturl(), ".")
 
@@ -28,7 +48,7 @@ def download(self, uri, timeout=None, headers={}, verify_ssl=True):
 
 
 class HTTPSHandler:
-    def __new__(self, verify_ssl=True):
+    def __new__(cls, verify_ssl: bool = True) -> urllib.request.HTTPSHandler:
         context = ssl.create_default_context()
         if not verify_ssl:
             context.check_hostname = False

m3u8/mixins.py

@@ -1,10 +1,19 @@
+from __future__ import annotations
+
+from collections.abc import Iterable
 from os.path import dirname
+from typing import Protocol
 from urllib.parse import urljoin, urlsplit
 
 
-class BasePathMixin:
+class HasUri(Protocol):
+    uri: str | None
+    base_uri: str | None
+
+
+class BasePathMixin(HasUri):
     @property
-    def absolute_uri(self):
+    def absolute_uri(self) -> str | None:
         if self.uri is None:
             return None
 
@@ -20,33 +29,40 @@ def absolute_uri(self):
         return ret
 
     @property
-    def base_path(self):
+    def base_path(self) -> str | None:
         if self.uri is None:
             return None
         return dirname(self.get_path_from_uri())
 
-    def get_path_from_uri(self):
+    def get_path_from_uri(self) -> str:
         """Some URIs have a slash in the query string."""
+        assert self.uri
         return self.uri.split("?")[0]
 
     @base_path.setter
-    def base_path(self, newbase_path):
-        if self.uri is not None:
-            if not self.base_path:
-                self.uri = f"{newbase_path}/{self.uri}"
-            else:
-                self.uri = self.uri.replace(self.base_path, newbase_path)
+    def base_path(self, newbase_path: str) -> None:
+        if self.uri is None:
+            return
+
+        if not self.base_path:
+            self.uri = f"{newbase_path}/{self.uri}"
+        else:
+            self.uri = self.uri.replace(self.base_path, newbase_path)
 
 
-class GroupedBasePathMixin:
-    def _set_base_uri(self, new_base_uri):
+class GroupedBasePathMixin[T: BasePathMixin](Iterable[T]):
+    @property
+    def base_uri(self) -> str: ...
+
+    @base_uri.setter
+    def base_uri(self, new_base_uri: str) -> None:
         for item in self:
             item.base_uri = new_base_uri
 
-    base_uri = property(None, _set_base_uri)
+    @property
+    def base_path(self) -> str: ...
 
-    def _set_base_path(self, newbase_path):
+    @base_path.setter
+    def base_path(self, newbase_path: str) -> None:
         for item in self:
             item.base_path = newbase_path
-
-    base_path = property(None, _set_base_path)