Maintenance: Add type hints across the board

Author: amotl

mqttwarn/commands.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# (c) 2014-2019 The mqttwarn developers
+# (c) 2014-2023 The mqttwarn developers
 from __future__ import print_function
 
 import codecs
@@ -8,6 +8,7 @@
 import os
 import signal
 import sys
+import typing as t
 
 from docopt import docopt
 
@@ -92,7 +93,13 @@ def run():
         run_mqttwarn()
 
 
-def launch_plugin_standalone(plugin, options, data, configfile=None, config_more=None):
+def launch_plugin_standalone(
+    plugin: str,
+    options: t.Dict,
+    data: t.Dict,
+    configfile: t.Optional[str] = None,
+    config_more: t.Optional[t.Dict] = None,
+):
 
     # Optionally load configuration file
     does_not_exist = False
@@ -148,7 +155,7 @@ def run_mqttwarn():
     subscribe_forever()
 
 
-def setup_logging(config):
+def setup_logging(config: Config):
     LOGLEVEL = config.loglevelnumber
     LOGFILE = config.logfile
     LOGFORMAT = config.logformat

mqttwarn/configuration.py

@@ -1,10 +1,11 @@
 # -*- coding: utf-8 -*-
-# (c) 2014-2022 The mqttwarn developers
+# (c) 2014-2023 The mqttwarn developers
 import ast
 import codecs
 import logging
 import os
 import sys
+import typing as t
 from configparser import NoOptionError, RawConfigParser
 
 from mqttwarn.util import load_functions
@@ -21,13 +22,13 @@
 
 class Config(RawConfigParser):
 
-    specials = {
+    specials: t.Dict[str, t.Union[bool, None]] = {
         "TRUE": True,
         "FALSE": False,
         "NONE": None,
     }
 
-    def __init__(self, configuration_file=None, defaults=None):
+    def __init__(self, configuration_file: t.Optional[str] = None, defaults: t.Optional[t.Dict] = None):
 
         defaults = defaults or {}
 
@@ -48,6 +49,7 @@ def __init__(self, configuration_file=None, defaults=None):
         self.password = None
         self.clientid = None
         self.lwt = None
+        self.lwt_alive = None
         self.skipretained = False
         self.cleansession = False
         self.protocol = 3
@@ -110,7 +112,7 @@ def __init__(self, configuration_file=None, defaults=None):
 
             self.functions = load_functions(functions_file)
 
-    def level2number(self, level):
+    def level2number(self, level: str) -> int:
 
         levels = {
             "CRITICAL": 50,
@@ -125,7 +127,8 @@ def level2number(self, level):
 
         return levels.get(level.upper(), levels["DEBUG"])
 
-    def g(self, section, key, default=None):
+    def g(self, section: str, key: str, default=None) -> t.Any:
+        val = None
         try:
             val = self.get(section, key)
             if isinstance(val, str) and val.upper() in self.specials:
@@ -139,29 +142,29 @@ def g(self, section, key, default=None):
             return val
         except:
             raise
-            return val
 
-    def getlist(self, section, key):
+    def getlist(self, section: str, key: str) -> t.Union[t.List, None]:
         """Return a list, fail if it isn't a list"""
 
         val = None
         try:
-            val = self.get(section, key)
-            val = [s.strip() for s in val.split(",")]
+            val_str = self.get(section, key)
+            val = [s.strip() for s in val_str.split(",")]
         except Exception as e:
             logger.warning("Expecting a list in section `%s', key `%s' (%s)" % (section, key, e))
 
         return val
 
-    def getdict(self, section, key):
+    # TODO: Add return type annotation.
+    def getdict(self, section: str, key: str):
         val = self.g(section, key)
 
         try:
             return dict(val)
         except:
             return None
 
-    def config(self, section):
+    def config(self, section: str) -> t.Dict:
         """Convert a whole section's options (except the options specified
         explicitly below) into a dict, turning
 
@@ -183,17 +186,17 @@ def config(self, section):
         return d
 
 
-def load_configuration(configfile=None, name="mqttwarn"):
+def load_configuration(configfile: t.Optional[str] = None, name: str = "mqttwarn") -> Config:
 
     if configfile is None:
-        configfile = os.getenv(name.upper() + "INI", name + ".ini")
+        configfile = str(os.getenv(name.upper() + "INI", name + ".ini"))
 
     if not os.path.exists(configfile):
         raise FileNotFoundError('Configuration file "{}" does not exist'.format(configfile))
 
     # TODO: There should be a factory method which creates a `Config` instance,
     #       including defaults, but without loading a configuration file.
-    defaults = {
+    defaults: t.Dict[str, str] = {
         "clientid": name,
         "lwt": "clients/{}".format(name),
         "lwt_alive": "1",

mqttwarn/context.py

@@ -1,11 +1,12 @@
 # -*- coding: utf-8 -*-
-# (c) 2014-2022 The mqttwarn developers
+# (c) 2014-2023 The mqttwarn developers
 import logging
 import typing as t
 
 import attr
 
 from mqttwarn.configuration import Config
+from mqttwarn.model import Service, TdataType, TopicTargetType
 from mqttwarn.util import load_function, sanitize_function_name
 
 logger = logging.getLogger(__name__)
@@ -20,9 +21,9 @@ class RuntimeContext:
     """
 
     config: Config = attr.ib()
-    invoker: t.Optional["FunctionInvoker"] = attr.ib()
+    invoker: "FunctionInvoker" = attr.ib()
 
-    def get_sections(self):
+    def get_sections(self) -> t.List[str]:
         sections = []
         for section in self.config.sections():
             if section == "defaults":
@@ -39,57 +40,57 @@ def get_sections(self):
                 logger.warning("Section `%s' has no targets defined" % section)
         return sections
 
-    def get_topic(self, section):
+    def get_topic(self, section: str) -> str:
         if self.config.has_option(section, "topic"):
             return self.config.get(section, "topic")
         return section
 
-    def get_qos(self, section):
+    def get_qos(self, section: str) -> int:
         qos = 0
         if self.config.has_option(section, "qos"):
             qos = int(self.config.get(section, "qos"))
         return qos
 
-    def get_config(self, section, name):
+    def get_config(self, section: str, name: str) -> t.Any:
         value = None
         if self.config.has_option(section, name):
             value = self.config.get(section, name)
         return value
 
-    def is_filtered(self, section, topic, payload):
+    def is_filtered(self, section: str, topic: str, payload: t.AnyStr) -> bool:
         if self.config.has_option(section, "filter"):
-            filterfunc = sanitize_function_name(self.config.get(section, "filter"))
             try:
-                return self.invoker.filter(filterfunc, topic, payload, section)
+                name = sanitize_function_name(self.config.get(section, "filter"))
+                return self.invoker.filter(name, topic, payload, section)
             except Exception as e:
-                logger.exception("Cannot invoke filter function '%s' defined in '%s': %s" % (filterfunc, section, e))
+                logger.exception("Cannot invoke filter function '%s' defined in '%s': %s" % (name, section, e))
         return False
 
-    def get_topic_data(self, section, topic):
+    def get_topic_data(self, section: str, data: TdataType) -> t.Optional[TdataType]:
         if self.config.has_option(section, "datamap"):
-            name = sanitize_function_name(self.config.get(section, "datamap"))
             try:
-                return self.invoker.datamap(name, topic)
+                name = sanitize_function_name(self.config.get(section, "datamap"))
+                return self.invoker.datamap(name, data)
             except Exception as e:
                 logger.exception("Cannot invoke datamap function '%s' defined in '%s': %s" % (name, section, e))
         return None
 
-    def get_all_data(self, section, topic, data):
+    def get_all_data(self, section: str, topic: str, data: TdataType) -> t.Optional[TdataType]:
         if self.config.has_option(section, "alldata"):
-            name = sanitize_function_name(self.config.get(section, "alldata"))
             try:
+                name = sanitize_function_name(self.config.get(section, "alldata"))
                 return self.invoker.alldata(name, topic, data)
             except Exception as e:
                 logger.exception("Cannot invoke alldata function '%s' defined in '%s': %s" % (name, section, e))
         return None
 
-    def get_topic_targets(self, section, topic, data):
+    def get_topic_targets(self, section: str, topic: str, data: TdataType) -> TopicTargetType:
         """
         Topic targets function invoker.
         """
         if self.config.has_option(section, "targets"):
-            name = sanitize_function_name(self.config.get(section, "targets"))
             try:
+                name = sanitize_function_name(self.config.get(section, "targets"))
                 return self.invoker.topic_target_list(name, topic, data)
             except Exception as ex:
                 error = repr(ex)
@@ -99,20 +100,31 @@ def get_topic_targets(self, section, topic, data):
                 )
         return None
 
-    def get_service_config(self, service):
+    def get_service_config(self, service: str) -> t.Dict[str, t.Any]:
         config = self.config.config("config:" + service)
         if config is None:
             return {}
         return dict(config)
 
-    def get_service_targets(self, service):
-        # Be more graceful with jobs w/o any target address information (2021-10-18 [amo]).
+    def get_service_targets(self, service: str) -> t.List[TopicTargetType]:
+        """
+        Resolve target address descriptor.
+
+        2021-10-18 [amo]: Be more graceful with jobs w/o any target address information.
+        """
+        targets: t.List[TopicTargetType] = []
         try:
-            targets = self.config.getdict("config:" + service, "targets") or [None]
-            return targets
+            targets = self.config.getdict("config:" + service, "targets")
         except:
             logger.exception("Unable to access targets for service `%s'" % service)
 
+        # TODO: The target address descriptor may be of any type these days,
+        #       and not necessarily a list.
+        # TODO: Currently, this makes sure to always return one element.
+        #       Verify if this is really needed.
+        targets = targets or [None]
+        return targets
+
 
 @attr.s
 class FunctionInvoker:
@@ -121,31 +133,32 @@ class FunctionInvoker:
     functions from a configured Python source code file.
     """
 
-    config = attr.ib()
-    srv = attr.ib()
+    config: Config = attr.ib()
+    srv: Service = attr.ib()
 
-    def datamap(self, name, topic):
+    def datamap(self, name: str, data: TdataType) -> TdataType:
         """
         Invoke function "name" loaded from the "functions" Python module.
 
         :param name:    Function name to invoke
-        :param topic:   Topic to pass to the invoked function
+        :param data:    Data to pass to the invoked function
         :return:        Return value of function invocation
         """
 
         val = None
+
         try:
             func = load_function(name=name, py_mod=self.config.functions)
             try:
-                val = func(topic, self.srv)  # new version
+                val = func(data, self.srv)  # new version
             except TypeError:
-                val = func(topic)  # legacy
+                val = func(data)  # legacy
         except:
             raise
 
         return val
 
-    def alldata(self, name, topic, data):
+    def alldata(self, name: str, topic: str, data: TdataType) -> TdataType:
         """
         Invoke function "name" loaded from the "functions" Python module.
 
@@ -164,7 +177,7 @@ def alldata(self, name, topic, data):
 
         return val
 
-    def topic_target_list(self, name, topic, data):
+    def topic_target_list(self, name: str, topic: str, data: TdataType) -> TopicTargetType:
         """
         Invoke function "name" loaded from the "functions" Python module.
         Computes dynamic topic subscription targets.
@@ -185,7 +198,7 @@ def topic_target_list(self, name, topic, data):
 
         return val
 
-    def filter(self, name, topic, payload, section=None):  # noqa:A003
+    def filter(self, name: str, topic: str, payload: t.AnyStr, section: t.Optional[str] = None) -> bool:  # noqa:A003
         """
         Invoke function "name" loaded from the "functions" Python module.
         Return that function's True/False.
@@ -199,15 +212,17 @@ def filter(self, name, topic, payload, section=None):  # noqa:A003
         # Filtering currently only works on text.
         # TODO: To let filtering also work on binary data, this line would need to go elsewhere. But where?
         if isinstance(payload, bytes):
-            payload = payload.decode("utf-8")
+            payload_decoded = payload.decode("utf-8")
+        else:
+            payload_decoded = payload
 
         rc = False
         try:
             func = load_function(name=name, py_mod=self.config.functions)
             try:
-                rc = func(topic, payload, section, self.srv)  # new version
+                rc = func(topic, payload_decoded, section, self.srv)  # new version
             except TypeError:
-                rc = func(topic, payload)  # legacy signature
+                rc = func(topic, payload_decoded)  # legacy signature
         except:
             raise