Big typing improvements

- Added 'Optional' to a bunch of None params.
- Started to enable mypy (pre-commit) but this will require considerable effort

.pre-commit-config.yaml

@@ -69,3 +69,20 @@ repos:
   #   rev: master
   #   hooks:
   #     - id: gitlint
+
+  # Yup this one is too busy too.  I attacked *some* of these, but there were still 270+ errors remaining to address (and some of these are like wack-a-mole, a small "fix" breaks something else.)
+  # - repo: https://github.com/pre-commit/mirrors-mypy
+  #   rev: v1.6.0
+  #   hooks:
+  #     - id: mypy
+  #       args:
+  #         - '--explicit-package-bases'
+  #       exclude: '(plugins/.*|splunk_app/.*)'
+  #       additional_dependencies:
+  #         - argcomplete
+  #         - importlib-metadata
+  #         - jinja2
+  #         - lxml
+  #         - pluggy
+  #         - pyyaml
+  #         - splunk-sdk

ksconf/app/deploy.py

@@ -9,7 +9,7 @@
 from enum import Enum
 from os import fspath
 from pathlib import Path, PurePath, PurePosixPath
-from typing import Set
+from typing import NewType, Optional, Set, Union
 
 from ksconf.app import AppManifest
 from ksconf.archive import extract_archive
@@ -39,21 +39,15 @@ def __str__(self):
 @dataclass
 class DeployAction:
     # Abstract base class
-    action: str
+    action: DeployActionType
 
     def to_dict(self) -> dict:
         data = asdict(self)
-        data["action"] = str(self.action)
-
-        for dc_field in fields(self):
-            if dc_field.name == "action":
-                continue
-
-            value = data[dc_field.name]
-            t = eval(dc_field.type)
-            if issubclass(t, PurePath):
-                value = fspath(value)
-                data[dc_field.name] = value
+        action = data.pop("action")
+        for key, value in data.items():
+            if isinstance(value, PurePath):
+                data[key] = fspath(value)
+        data["action"] = str(action)
         return data
 
     @classmethod
@@ -66,44 +60,46 @@ def from_dict(self, data: dict) -> DeployAction:
             if dc_field.name == "action":
                 continue
             value = data[dc_field.name]
-            t = eval(dc_field.type)
-            if issubclass(t, PurePath):
-                value = t(value)
+            if "PurePosixPath" in dc_field.type:
+                value = PurePosixPath(value)
                 data[dc_field.name] = value
         return da_class(**data)
 
 
+DeployAction_T = NewType("DeployAction_T", DeployAction)
+
+
 @dataclass
 class DeployAction_ExtractFile(DeployAction):
-    action: str = field(init=False, default=DeployActionType.EXTRACT_FILE)
+    action: DeployActionType = field(init=False, default=DeployActionType.EXTRACT_FILE)
     subtype: str
     path: PurePosixPath
-    mode: int = None
-    mtime: int = None
-    hash: str = None
-    rel_path: str = None
+    mode: Optional[int] = None
+    mtime: Optional[int] = None
+    hash: Optional[str] = None
+    rel_path: Optional[str] = None
 
 
 @dataclass
 class DeployAction_RemoveFile(DeployAction):
-    action: str = field(init=False, default=DeployActionType.REMOVE_FILE)
+    action: DeployActionType = field(init=False, default=DeployActionType.REMOVE_FILE)
     path: PurePosixPath
 
 
 @dataclass
 class DeployAction_SetAppName(DeployAction):
-    action: str = field(init=False, default=DeployActionType.SET_APP_NAME)
+    action: DeployActionType = field(init=False, default=DeployActionType.SET_APP_NAME)
     name: str
 
 
 @dataclass
 class DeployAction_SourceReference(DeployAction):
-    action: str = field(init=False, default=DeployActionType.SOURCE_REFERENCE)
+    action: DeployActionType = field(init=False, default=DeployActionType.SOURCE_REFERENCE)
     archive_path: str
     hash: str
 
 
-def get_deploy_action_class(action: str) -> DeployAction:
+def get_deploy_action_class(action: Union[DeployActionType, str]) -> DeployAction_T:
     return {
         DeployActionType.EXTRACT_FILE: DeployAction_ExtractFile,
         DeployActionType.REMOVE_FILE: DeployAction_RemoveFile,
@@ -139,7 +135,7 @@ def __init__(self):
         self.actions: List[DeployAction] = []
         self.actions_by_type = Counter()
 
-    def add(self, action: str, *args, **kwargs):
+    def add(self, action: Union[str, DeployAction], *args, **kwargs):
         if not isinstance(action, DeployAction):
             action_cls = get_deploy_action_class(action)
             action = action_cls(*args, **kwargs)

ksconf/app/facts.py

@@ -10,41 +10,46 @@
 from dataclasses import asdict, dataclass, field, fields
 from os import fspath
 from pathlib import Path
-from typing import ClassVar
+from typing import Any, ClassVar, Optional
 
 from ksconf.app.manifest import AppArchiveContentError
 from ksconf.archive import extract_archive, gaf_filter_name_like
-from ksconf.compat import List, Tuple
+from ksconf.compat import Dict, List, Set, Tuple
 from ksconf.conf.merge import merge_conf_dicts
 from ksconf.conf.parser import (PARSECONF_LOOSE, ConfType, conf_attr_boolean,
                                 default_encoding, parse_conf, parse_string)
 
+OInt = Optional[int]
+OStr = Optional[str]
+OBool = Optional[bool]
+
 
 @dataclass
 class AppFacts:
     """ Basic Splunk application info container.
     A majority of these facts are extracted from ``app.conf``
     """
     name: str
-    label: str = None
-    id: str = None
-    version: str = None
-    author: str = None
-    description: str = None
-    state: str = None
-    build: int = None
-
-    is_configured: bool = field(init=False, default=None)
-    allows_disable: bool = field(init=False, default=None)
-    state_change_requires_restart: bool = field(init=False, default=None)
-
-    install_source_checksum: str = field(init=False, default=None)
-    install_source_local_checksum: str = field(init=False, default=None)
-    check_for_updates: bool = field(init=False, default=None)
-    is_visible: bool = field(init=False, default=None)
-
-    deployer_lookups_push_mode: str = field(init=False, default=None)
-    deployer_push_mode: str = field(init=False, default=None)
+
+    label: OStr = None
+    id: OStr = None
+    version: OStr = None
+    author: OStr = None
+    description: OStr = None
+    state: OStr = None
+    build: OStr = None
+
+    is_configured: OBool = field(init=False, default=None)
+    allows_disable: OBool = field(init=False, default=None)
+    state_change_requires_restart: OBool = field(init=False, default=None)
+
+    install_source_checksum: OStr = field(init=False, default=None)
+    install_source_local_checksum: OStr = field(init=False, default=None)
+    check_for_updates: OBool = field(init=False, default=None)
+    is_visible: OBool = field(init=False, default=None)
+
+    deployer_lookups_push_mode: OStr = field(init=False, default=None)
+    deployer_push_mode: OStr = field(init=False, default=None)
 
     _conf_translate_pairs: ClassVar[List[Tuple[str, List[str]]]] = [
         ("launcher", [
@@ -75,26 +80,28 @@ class AppFacts:
     def to_dict(self) -> dict:
         return asdict(self)
 
-    def to_tiny_dict(self, *keep_attrs) -> dict:
+    def to_tiny_dict(self, *keep_attrs: str) -> Dict[str, Any]:
         """ Return dict representation, discarding the Nones """
         return {k: v for k, v in asdict(self).items() if v is not None or k in keep_attrs}
 
     @classmethod
-    def from_conf(cls, name, conf: ConfType) -> AppFacts:
+    def from_conf(cls, name: str, conf: ConfType) -> AppFacts:
         """
         Create AppFacts from an app.conf configuration content.
         """
         new = cls(name)
 
-        type_mapping = {f.name: f.type for f in fields(cls) if f.type in ("bool", "int")}
+        # Another possible option:
+        # typing.get_type_hints(ksconf.app.facts.AppFacts)["build"].__args__[0]      (__args__[0] due to Optional)
+        type_mapping = {f.name: f.type for f in fields(cls) if f.type in ("OBool", "OInt")}
 
         def convert_attr(attr_name, value):
             # XXX: Is there a better approach for dataclasses?  fields() returns a list
             data_type = type_mapping.get(attr_name, None)
             convert_function = None
-            if data_type == "bool":
+            if data_type == "OBool":
                 convert_function = conf_attr_boolean
-            elif data_type == "int":
+            elif data_type == "OInt":
                 convert_function = int
             else:
                 return value
@@ -135,8 +142,8 @@ def from_archive(cls, archive: Path):
         ''' Returns list of app names, merged app_conf and a dictionary of extra facts that may be useful '''
         archive = Path(archive)
 
-        app_names = set()
-        app_confs = defaultdict(dict)
+        app_names: Set[str] = set()
+        app_confs: Dict[str, str] = defaultdict(dict)
 
         is_app_conf = gaf_filter_name_like("app.conf")
         for gaf in extract_archive(archive, extract_filter=is_app_conf):

ksconf/app/manifest.py

@@ -10,7 +10,7 @@
 from dataclasses import asdict, dataclass, field
 from os import fspath
 from pathlib import Path, PurePosixPath
-from typing import Callable, Iterable
+from typing import Callable, Iterable, Optional
 
 from ksconf.archive import extract_archive
 from ksconf.compat import List
@@ -45,9 +45,9 @@ class AppManifestFile:
     path: PurePosixPath
     mode: int
     size: int
-    hash: str = None
+    hash: Optional[str] = None
 
-    def content_match(self, other):
+    def content_match(self, other: AppManifestFile):
         return self.hash == other.hash
 
     def to_dict(self):
@@ -88,8 +88,8 @@ class AppManifest:
     * :py:meth:`from_filesystem` - from an extracted Splunk app directory
     * :py:meth:`from_dict` - primarily for json serialization from :py:meth:`to_dict`.
     """
-    name: str = None
-    source: str = None
+    name: Optional[str] = None
+    source: Optional[str] = None
     hash_algorithm: str = field(default=MANIFEST_HASH)
     _hash: str = field(default=UNSET, init=False)
     files: List[AppManifestFile] = field(default_factory=list)
@@ -162,7 +162,7 @@ def to_dict(self):
     def from_archive(cls, archive: Path,
                      calculate_hash=True,
                      *,
-                     filter_file: Callable = None) -> AppManifest:
+                     filter_file: Optional[Callable] = None) -> AppManifest:
         """
         Create as new AppManifest from a tarball.  Set ``calculate_hash`` as
         False when only a file listing is needed.
@@ -198,11 +198,11 @@ def gethash(_):
 
     @classmethod
     def from_filesystem(cls, path: Path,
-                        name: str = None,
+                        name: Optional[str] = None,
                         follow_symlinks=False,
                         calculate_hash=True,
                         *,
-                        filter_file: Callable = None) -> AppManifest:
+                        filter_file: Optional[Callable] = None) -> AppManifest:
         """
         Create as new AppManifest from an existing directory structure.
         Set ``calculate_hash`` as False when only a file listing is needed.
@@ -318,7 +318,7 @@ def from_json_manifest(cls,
                            archive: Path,
                            stored_file: Path,
                            *,
-                           permanent_archive: Path = None) -> StoredArchiveManifest:
+                           permanent_archive: Optional[Path] = None) -> StoredArchiveManifest:
         """
         Attempt to load an archive stored manifest from ``archive`` and ``stored_file`` paths.
         If the archive has changed since the manifest was stored, then an
@@ -370,11 +370,11 @@ def create_manifest_from_archive(
 
 def load_manifest_for_archive(
         archive: Path,
-        manifest_file: Path = None,
+        manifest_file: Optional[Path] = None,
         *,
         read_manifest=True,
         write_manifest=True,
-        permanent_archive: Path = None,
+        permanent_archive: Optional[Path] = None,
         log_callback=print) -> AppManifest:
     """
     Load manifest for ``archive`` and create a stored copy of the manifest in

ksconf/archive.py

@@ -2,7 +2,7 @@
 
 import os
 from fnmatch import fnmatch
-from typing import Iterable, NamedTuple, Sequence, Tuple, Union
+from typing import ByteString, Callable, Iterable, NamedTuple, Optional, Sequence, Tuple
 
 from ksconf.consts import RegexType
 
@@ -11,10 +11,10 @@ class GenArchFile(NamedTuple):
     path: str
     mode: int
     size: int
-    payload: Union[bytes, None]
+    payload: Optional[ByteString]
 
 
-def extract_archive(archive_name, extract_filter: callable = None) -> Iterable[GenArchFile]:
+def extract_archive(archive_name, extract_filter: Optional[Callable] = None) -> Iterable[GenArchFile]:
     if extract_filter is not None and not callable(extract_filter):  # pragma: no cover
         raise ValueError("extract_filter must be a callable!")
     archive_name = os.fspath(archive_name)

ksconf/builder/__init__.py

@@ -6,7 +6,7 @@
 import sys
 from pathlib import Path
 from subprocess import Popen
-from typing import Callable, List, TextIO
+from typing import Callable, List, Optional, TextIO
 
 from ksconf.consts import EXIT_CODE_INTERNAL_ERROR, is_debug
 
@@ -28,8 +28,8 @@ class BuildStep:
 
     def __init__(self,
                  build: Path,
-                 source: Path = None,
-                 dist: Path = None,
+                 source: Optional[Path] = None,
+                 dist: Optional[Path] = None,
                  output: TextIO = sys.stdout):
         self.build_path = build
         self.source_path = source
@@ -54,7 +54,7 @@ def is_quiet(self):
     def is_verbose(self):
         return self.verbosity >= VERBOSE
 
-    def get_logger(self, prefix: str = None) -> Callable:
+    def get_logger(self, prefix: Optional[str] = None) -> Callable:
         if prefix is None:
             prefix = inspect.currentframe().f_back.f_code.co_name
         elif re.match(r'[\w_]+', prefix):