From: Kaliko Jack <kaliko@azylum.org>
Date: Fri, 31 Mar 2023 14:49:41 +0000 (+0200)
Subject: First shot at type hints
X-Git-Url: https://git.kaliko.me/?a=commitdiff_plain;h=69bf5d68cc66285f01c7d0ef08cb852d023cf5aa;p=python-musicpd.git

First shot at type hints
---

diff --git a/musicpd.py b/musicpd.py
index e2d1265..61b8b6f 100644
--- a/musicpd.py
+++ b/musicpd.py
@@ -11,6 +11,9 @@ import socket
 import os
 
 from functools import wraps
+# Type hint for python <= 3.8
+from typing import Any, Dict, List, Tuple
+from typing import Literal, Optional, Union
 
 HELLO_PREFIX = "OK MPD "
 ERROR_PREFIX = "ACK "
@@ -19,9 +22,9 @@ NEXT = "list_OK"
 VERSION = '0.9.0b0'
 #: Seconds before a connection attempt times out
 #: (overriden by MPD_TIMEOUT env. var.)
-CONNECTION_TIMEOUT = 30
+CONNECTION_TIMEOUT: int = 30
 #: Socket timeout in second (Default is None for no timeout)
-SOCKET_TIMEOUT = None
+SOCKET_TIMEOUT: Union[int, None] = None
 
 
 def iterator_wrapper(func):
@@ -73,7 +76,7 @@ class IteratingError(MPDError):
 
 class Range:
 
-    def __init__(self, tpl):
+    def __init__(self, tpl: Tuple[int]):
         self.tpl = tpl
         self._check()
 
@@ -145,13 +148,13 @@ class MPDClient:
       will not be used as default value for the :py:meth:`password` method
     """
 
-    def __init__(self):
-        self.iterate = False
+    def __init__(self) -> None:
+        self.iterate: bool = False
         #: Socket timeout value in seconds
         self._socket_timeout = SOCKET_TIMEOUT
         #: Current connection timeout value, defaults to
         #: :py:obj:`CONNECTION_TIMEOUT` or env. var. ``MPD_TIMEOUT`` if provided
-        self.mpd_timeout = None
+        self.mpd_timeout: Union[None,int] = None
         self._reset()
         self._commands = {
             # Status Commands
@@ -286,19 +289,20 @@ class MPDClient:
         }
         self._get_envvars()
 
-    def _get_envvars(self):
+    def _get_envvars(self) -> None:
         """
         Retrieve MPD env. var. to overrides "localhost:6600"
             Use MPD_HOST/MPD_PORT if set
             else use MPD_HOST=${XDG_RUNTIME_DIR:-/run/}/mpd/socket if file exists
         """
-        self.host = 'localhost'
-        self.pwd = None
-        self.port = os.getenv('MPD_PORT', '6600')
-        if os.getenv('MPD_HOST'):
+        self.host: str = 'localhost'
+        self.pwd: Union[None, str] = None
+        self.port: Union[int,str] = os.getenv('MPD_PORT', '6600')
+        _host: str = os.getenv('MPD_HOST', '')
+        if _host:
             # If password is set: MPD_HOST=pass@host
-            if '@' in os.getenv('MPD_HOST'):
-                mpd_host_env = os.getenv('MPD_HOST').split('@', 1)
+            if '@' in _host:
+                mpd_host_env = _host.split('@', 1)
                 if mpd_host_env[0]:
                     # A password is actually set
                     self.pwd = mpd_host_env[0]
@@ -309,16 +313,16 @@ class MPDClient:
                     self.host = '@'+mpd_host_env[1]
             else:
                 # MPD_HOST is a plain host
-                self.host = os.getenv('MPD_HOST')
+                self.host = _host
         else:
             # Is socket there
             xdg_runtime_dir = os.getenv('XDG_RUNTIME_DIR', '/run')
             rundir = os.path.join(xdg_runtime_dir, 'mpd/socket')
             if os.path.exists(rundir):
                 self.host = rundir
-        self.mpd_timeout = os.getenv('MPD_TIMEOUT')
-        if self.mpd_timeout and self.mpd_timeout.isdigit():
-            self.mpd_timeout = int(self.mpd_timeout)
+        _mpd_timeout = os.getenv('MPD_TIMEOUT', 'X')
+        if _mpd_timeout.isdigit():
+            self.mpd_timeout = int(_mpd_timeout)
         else:  # Use CONNECTION_TIMEOUT as default even if MPD_TIMEOUT carries gargage
             self.mpd_timeout = CONNECTION_TIMEOUT
 
@@ -412,7 +416,7 @@ class MPDClient:
             amount -= len(result)
         return bytes(chunk)
 
-    def _read_line(self, binary=False):
+    def _read_line(self, binary: Literal[True,False] = False):
         if binary:
             line = self._rbfile.readline().decode('utf-8')
         else:
@@ -433,7 +437,7 @@ class MPDClient:
             return None
         return line
 
-    def _read_pair(self, separator, binary=False):
+    def _read_pair(self, separator: str, binary: Literal[True,False] = False):
         line = self._read_line(binary=binary)
         if line is None:
             return None
@@ -442,7 +446,7 @@ class MPDClient:
             raise ProtocolError(f"Could not parse pair: '{line}'")
         return pair
 
-    def _read_pairs(self, separator=": ", binary=False):
+    def _read_pairs(self, separator=": ", binary: Literal[True,False] =False):
         pair = self._read_pair(separator, binary=binary)
         while pair:
             yield pair
@@ -461,8 +465,8 @@ class MPDClient:
         for _, value in self._read_pairs(":"):
             yield value
 
-    def _read_objects(self, delimiters=None):
-        obj = {}
+    def _read_objects(self, delimiters: Optional[List[str]] = None):
+        obj: Dict[str,Any] = {}
         if delimiters is None:
             delimiters = []
         for key, value in self._read_pairs():
@@ -639,7 +643,7 @@ class MPDClient:
         self._write_command("noidle")
         return self._fetch_list()
 
-    def connect(self, host=None, port=None):
+    def connect(self, host: Optional[str] = None, port: Optional[Union[int, str]] = None):
         """Connects the MPD server
 
         :param str host: hostname, IP or FQDN (defaults to `localhost` or socket, see below for details)
@@ -750,7 +754,7 @@ class MPDClient:
         return self._fetch_command_list()
 
 
-def escape(text):
+def escape(text: str) -> str:
     return text.replace("\\", "\\\\").replace('"', '\\"')
 
 # vim: set expandtab shiftwidth=4 softtabstop=4 textwidth=79: