Minor flake8 & typing enhancements

ksconf/commands/restpublish.py

@@ -14,15 +14,15 @@
 
 import os
 import sys
-from argparse import ArgumentParser
+from argparse import ArgumentParser, Namespace
 from urllib.parse import urlparse
 
 from ksconf.commands import (ConfFileProxy, ConfFileType, KsconfCmd,
                              add_splunkd_access_args, add_splunkd_namespace,
                              dedent)
 from ksconf.conf.delta import DiffHeader, compare_stanzas, is_equal, reduce_stanza, show_diff
 from ksconf.conf.meta import MetaData
-from ksconf.conf.parser import GLOBAL_STANZA, PARSECONF_LOOSE, conf_attr_boolean
+from ksconf.conf.parser import GLOBAL_STANZA, PARSECONF_LOOSE, ConfType, conf_attr_boolean
 from ksconf.consts import EXIT_CODE_SUCCESS
 from ksconf.util.completers import conf_files_completer
 
@@ -106,11 +106,11 @@ def register_args(self, parser: ArgumentParser):
             """))
 
     @staticmethod
-    def make_boolean(stanza, attr="disabled"):
+    def make_boolean(stanza: ConfType, attr: str = "disabled"):
         if attr in stanza:
             stanza[attr] = "1" if conf_attr_boolean(stanza[attr]) else "0"
 
-    def connect_splunkd(self, args):
+    def connect_splunkd(self, args: Namespace):
         up = urlparse(args.url)
         # Take username/password form URL, if encoded there; otherwise use defaults from argparse
         if args.session_key:
@@ -140,13 +140,12 @@ def connect_splunkd(self, args):
                              f"{login_fail_info}:  {e}\n")
             raise e
 
-    def handle_conf_file(self, args, conf_proxy):
+    def handle_conf_file(self, args: Namespace, conf_proxy: ConfFileProxy):
         if args.conf_type:
             conf_type = args.conf_type
         else:
             conf_type = os.path.basename(conf_proxy.name).replace(".conf", "")
 
-
         try:
             config_file = self._service.confs[conf_type]
         except KeyError:
@@ -191,7 +190,10 @@ def handle_conf_file(self, args, conf_proxy):
             if "acl_delta" in info:
                 show_diff(self.stdout, info["acl_delta"])
 
-    def publish_conf(self, stanza_name, stanza_data, config_file):
+    def publish_conf(self,
+                     stanza_name: str,
+                     stanza_data: ConfType,
+                     config_file):
         if self.meta:
             metadata = self.meta.get(config_file.name, stanza_name)
             owner = metadata.get("owner", None)
@@ -324,7 +326,10 @@ def publish_conf(self, stanza_name, stanza_data, config_file):
 
         return (action, res)
 
-    def delete_conf(self, stanza_name, stanza_data, config_file):
+    def delete_conf(self,
+                    stanza_name: str,
+                    stanza_data: ConfType,
+                    config_file):
         res = {}
         if stanza_name in config_file:
             stz = config_file[stanza_name]
@@ -346,7 +351,7 @@ def delete_conf(self, stanza_name, stanza_data, config_file):
             res["delta"] = []
             return ("nochange", res)
 
-    def run(self, args):
+    def run(self, args: Namespace):
         if args.insecure:
             raise NotImplementedError("Need to implement -k feature")
 

ksconf/conf/merge.py

@@ -46,7 +46,7 @@ def _merge_conf_dicts(base, new_layer):
     # Nothing to return, base is updated in-place
 
 
-def merge_conf_dicts(*dicts):
+def merge_conf_dicts(*dicts: ConfType) -> ConfType:
     result = {}
     for d in dicts:
         d = deepcopy(d)