Fix logic bugs and add comprehensive type hints

resources/lib/kodiUtilities.py

@@ -7,6 +7,7 @@
 import json
 import re
 import logging
+from typing import Tuple, List, Dict, Union, Optional
 from resources.lib import utilities
 
 
@@ -19,46 +20,46 @@
 
 
 def notification(
-    header: str, message: str, time=5000, icon=__addon__.getAddonInfo("icon")
-):
+    header: str, message: str, time: int = 5000, icon: str = __addon__.getAddonInfo("icon")
+) -> None:
     xbmcgui.Dialog().notification(header, message, icon, time)
 
 
-def showSettings():
+def showSettings() -> None:
     __addon__.openSettings()
 
 
-def getSetting(setting):
+def getSetting(setting: str) -> str:
     return __addon__.getSetting(setting).strip()
 
 
-def setSetting(setting, value):
+def setSetting(setting: str, value: Union[str, int, float, bool]) -> None:
     __addon__.setSetting(setting, str(value))
 
 
-def getSettingAsBool(setting):
+def getSettingAsBool(setting: str) -> bool:
     return getSetting(setting).lower() == "true"
 
 
-def getSettingAsFloat(setting):
+def getSettingAsFloat(setting: str) -> float:
     try:
         return float(getSetting(setting))
     except ValueError:
         return 0
 
 
-def getSettingAsInt(setting):
+def getSettingAsInt(setting: str) -> int:
     try:
         return int(getSettingAsFloat(setting))
     except ValueError:
         return 0
 
 
-def getString(string_id):
+def getString(string_id: int) -> str:
     return __addon__.getLocalizedString(string_id)
 
 
-def kodiJsonRequest(params):
+def kodiJsonRequest(params: Dict) -> Optional[Union[Dict, List]]:
     data = json.dumps(params)
     request = xbmc.executeJSONRPC(data)
 
@@ -131,7 +132,7 @@ def checkExclusion(fullpath: str) -> bool:
     return found
 
 
-def kodiRpcToTraktMediaObject(type, data, mode="collected"):
+def kodiRpcToTraktMediaObject(type: str, data: Dict, mode: str = "collected") -> Optional[Dict]:
     if type == "show":
         if "uniqueid" in data:
             data["ids"] = data.pop("uniqueid")
@@ -230,7 +231,7 @@ def kodiRpcToTraktMediaObject(type, data, mode="collected"):
         return
 
 
-def kodiRpcToTraktMediaObjects(data, mode="collected"):
+def kodiRpcToTraktMediaObjects(data: Dict, mode: str = "collected") -> Optional[List]:
     if "tvshows" in data:
         shows = data["tvshows"]
 
@@ -270,7 +271,7 @@ def kodiRpcToTraktMediaObjects(data, mode="collected"):
         return
 
 
-def getShowDetailsFromKodi(showID, fields):
+def getShowDetailsFromKodi(showID: int, fields: List) -> Optional[Dict]:
     result = kodiJsonRequest(
         {
             "jsonrpc": "2.0",
@@ -292,7 +293,7 @@ def getShowDetailsFromKodi(showID, fields):
         return None
 
 
-def getSeasonDetailsFromKodi(seasonID, fields):
+def getSeasonDetailsFromKodi(seasonID: int, fields: List) -> Optional[Dict]:
     result = kodiJsonRequest(
         {
             "jsonrpc": "2.0",
@@ -317,7 +318,7 @@ def getSeasonDetailsFromKodi(seasonID, fields):
 # get a single episode from kodi given the id
 
 
-def getEpisodeDetailsFromKodi(libraryId, fields):
+def getEpisodeDetailsFromKodi(libraryId: int, fields: List) -> Optional[Dict]:
     result = kodiJsonRequest(
         {
             "jsonrpc": "2.0",
@@ -358,7 +359,7 @@ def getEpisodeDetailsFromKodi(libraryId, fields):
 # get a single movie from kodi given the id
 
 
-def getMovieDetailsFromKodi(libraryId, fields):
+def getMovieDetailsFromKodi(libraryId: int, fields: List) -> Optional[Dict]:
     result = kodiJsonRequest(
         {
             "jsonrpc": "2.0",
@@ -380,7 +381,7 @@ def getMovieDetailsFromKodi(libraryId, fields):
         return None
 
 
-def checkAndConfigureProxy():
+def checkAndConfigureProxy() -> Optional[str]:
     proxyActive = kodiJsonRequest(
         {
             "jsonrpc": "2.0",
@@ -480,7 +481,7 @@ def checkAndConfigureProxy():
     return None
 
 
-def getMediaType():
+def getMediaType() -> Optional[str]:
     listType = xbmc.getInfoLabel("ListItem.DBTYPE")
 
     xbmc.log("list item type: %s" % listType, xbmc.LOGINFO)
@@ -497,7 +498,7 @@ def getMediaType():
         return None
 
 
-def getInfoLabelDetails(result):
+def getInfoLabelDetails(result: Dict) -> Tuple[str, Dict]:
     type = result["item"]["type"]
     data = {"action": "started"}
     # check type of item

resources/lib/service.py

@@ -433,7 +433,7 @@ def __init__(self, *args, **kwargs):
 
     def onNotification(self, sender, method, data):
         # method looks like Other.NEXTUPWATCHEDSIGNAL
-        if method.split(".")[1].upper() != "NEXTUPWATCHEDSIGNAL":
+        if "." not in method or method.split(".")[1].upper() != "NEXTUPWATCHEDSIGNAL":
             return
 
         logger.debug("Callback received - Upnext skipped to the next episode")

resources/lib/sqlitequeue.py

@@ -14,6 +14,7 @@
 import xbmcvfs
 import xbmcaddon
 import logging
+from typing import Any, Iterator, Optional
 
 logger = logging.getLogger(__name__)
 
@@ -59,27 +60,27 @@ def __len__(self) -> int:
             executed = conn.execute(self._count).fetchone()[0]
         return executed
 
-    def __iter__(self):
+    def __iter__(self) -> Iterator[Any]:
         with self._get_conn() as conn:
             for _, obj_buffer in conn.execute(self._iterate):
-                yield loads(str(obj_buffer))
+                yield loads(obj_buffer)
 
-    def _get_conn(self):
+    def _get_conn(self) -> sqlite3.Connection:
         id = get_ident()
         if id not in self._connection_cache:
             self._connection_cache[id] = sqlite3.Connection(self.path, timeout=60)
         return self._connection_cache[id]
 
-    def purge(self):
+    def purge(self) -> None:
         with self._get_conn() as conn:
             conn.execute(self._purge)
 
-    def append(self, obj):
+    def append(self, obj: Any) -> None:
         obj_buffer = dumps(obj)
         with self._get_conn() as conn:
             conn.execute(self._append, (obj_buffer,))
 
-    def get(self, sleep_wait=True):
+    def get(self, sleep_wait: bool = True) -> Optional[Any]:
         keep_pooling = True
         wait = 0.1
         max_wait = 2
@@ -89,10 +90,11 @@ def get(self, sleep_wait=True):
             while keep_pooling:
                 conn.execute(self._write_lock)
                 cursor = conn.execute(self._get)
-                try:
-                    id, obj_buffer = cursor.fetchone()
+                row = cursor.fetchone()
+                if row:
+                    id, obj_buffer = row
                     keep_pooling = False
-                except StopIteration:
+                else:
                     conn.commit()  # unlock the database
                     if not sleep_wait:
                         keep_pooling = False
@@ -102,13 +104,13 @@ def get(self, sleep_wait=True):
                     wait = min(max_wait, tries / 10 + wait)
             if id:
                 conn.execute(self._del, (id,))
-                return loads(str(obj_buffer))
+                return loads(obj_buffer)
         return None
 
-    def peek(self):
+    def peek(self) -> Optional[Any]:
         with self._get_conn() as conn:
             cursor = conn.execute(self._peek)
-            try:
-                return loads(str(cursor[0]))
-            except StopIteration:
-                return None
+            row = cursor.fetchone()
+            if row:
+                return loads(row[0])
+            return None