Prs/add mypy to pre commit checks

.pre-commit-config.yaml

@@ -15,3 +15,10 @@ repos:
       - id: mixed-line-ending
       - id: end-of-file-fixer
       - id: trailing-whitespace
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: 'v1.9.0'
+    hooks:
+    -   id: mypy
+        args: [--strict, --ignore-missing-imports, --check-untyped-defs]
+        additional_dependencies:
+          - types-PyYAML

setup.py

@@ -16,14 +16,15 @@
 limitations under the License.
 """
 
+from typing import Dict, Any
+
 try:
     from setuptools import find_packages, setup
 except ImportError:
-    from distutils import find_packages, setup
-#
+    from distutils import find_packages, setup  # type: ignore[attr-defined, unused-ignore]
 
 
-def get_version():
+def get_version() -> str:
     """
     Returns the version currently in development.
 
@@ -36,7 +37,7 @@ def get_version():
 
 #
 
-_setup = {
+_setup: Dict[str, Any] = {
     "version": get_version()[1:],
     "data_files": [("docs", ["LICENSE", "README.md"])],
     "test_suite": "tests",

src/rookify/__main__.py

@@ -6,7 +6,7 @@
 from .yaml import load_yaml, save_yaml
 
 
-def main():
+def main() -> None:
     try:
         config = load_yaml("config.yaml")
     except FileNotFoundError as err:

src/rookify/modules/__init__.py

@@ -2,6 +2,7 @@
 
 import importlib
 import types
+from typing import List
 
 from collections import OrderedDict
 from .module import ModuleHandler
@@ -23,7 +24,9 @@ def __init__(self, module_name: str, message: str):
         self.message = message
 
 
-def load_modules(module_names: list) -> tuple[list, list]:
+def load_modules(
+    module_names: List[str],
+) -> tuple[List[types.ModuleType], List[types.ModuleType]]:
     """
     Dynamically loads modules from the 'modules' package.
 
@@ -32,7 +35,7 @@ def load_modules(module_names: list) -> tuple[list, list]:
     """
 
     # Sanity checks for modules
-    def check_module_sanity(module_name: str, module: types.ModuleType):
+    def check_module_sanity(module_name: str, module: types.ModuleType) -> None:
         for attr_type, attr_name in (
             (ModuleHandler, "HANDLER_CLASS"),
             (str, "MODULE_NAME"),
@@ -52,9 +55,11 @@ def check_module_sanity(module_name: str, module: types.ModuleType):
                 )
 
     # Load the modules in the given list and recursivley load required modules
-    required_modules = OrderedDict()
+    required_modules: OrderedDict[str, types.ModuleType] = OrderedDict()
 
-    def load_required_modules(modules_out: OrderedDict, module_names: list) -> None:
+    def load_required_modules(
+        modules_out: OrderedDict[str, types.ModuleType], module_names: List[str]
+    ) -> None:
         for module_name in module_names:
             if module_name in modules_out:
                 continue
@@ -70,10 +75,12 @@ def load_required_modules(modules_out: OrderedDict, module_names: list) -> None:
     load_required_modules(required_modules, module_names)
 
     # Recursively load the modules in the PREFLIGHT_REQUIRES attribute of the given modules
-    preflight_modules = OrderedDict()
+    preflight_modules: OrderedDict[str, types.ModuleType] = OrderedDict()
 
     def load_preflight_modules(
-        modules_in: OrderedDict, modules_out: OrderedDict, module_names: list
+        modules_in: OrderedDict[str, types.ModuleType],
+        modules_out: OrderedDict[str, types.ModuleType],
+        module_names: List[str],
     ) -> None:
         for module_name in module_names:
             if module_name in modules_out:
@@ -94,13 +101,17 @@ def load_preflight_modules(
             if module_name not in modules_in:
                 modules_out[module_name] = module
 
-    load_preflight_modules(required_modules, preflight_modules, required_modules.keys())
+    load_preflight_modules(
+        required_modules, preflight_modules, list(required_modules.keys())
+    )
 
     # Sort the modules by the AFTER keyword
-    modules = OrderedDict()
+    modules: OrderedDict[str, types.ModuleType] = OrderedDict()
 
     def sort_modules(
-        modules_in: OrderedDict, modules_out: OrderedDict, module_names: list
+        modules_in: OrderedDict[str, types.ModuleType],
+        modules_out: OrderedDict[str, types.ModuleType],
+        module_names: List[str],
     ) -> None:
         for module_name in module_names:
             if module_name not in modules_in:

src/rookify/modules/analyze_ceph/__init__.py

@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
+# type: ignore
 
 from .main import AnalyzeCephHandler
 

src/rookify/modules/analyze_ceph/main.py

@@ -1,14 +1,15 @@
 # -*- coding: utf-8 -*-
 
-
 from ..module import ModuleHandler
 
+from typing import Any, Dict
+
 
 class AnalyzeCephHandler(ModuleHandler):
-    def run(self) -> dict:
+    def run(self) -> Dict[str, Any]:
         commands = ["mon dump", "osd dump", "device ls", "fs dump", "node ls"]
 
-        results = dict()
+        results: Dict[str, Any] = dict()
         for command in commands:
             parts = command.split(" ")
             leaf = results

src/rookify/modules/example/__init__.py

@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
+# type: ignore
 
 from .main import ExampleHandler
 

src/rookify/modules/example/main.py

@@ -2,12 +2,14 @@
 
 from ..module import ModuleHandler, ModuleException
 
+from typing import Any, Dict
+
 
 class ExampleHandler(ModuleHandler):
-    def preflight_check(self):
+    def preflight_check(self) -> None:
         # Do something for checking if all needed preconditions are met else throw ModuleException
         raise ModuleException("Example module was loaded, so aborting!")
 
-    def run(self) -> dict:
+    def run(self) -> Dict[str, Any]:
         # Run the migration tasks
-        pass
+        return {}

src/rookify/modules/migrate_monitors/__init__.py

@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
+# type: ignore
 
 from .main import MigrateMonitorsHandler
 

src/rookify/modules/migrate_osds/__init__.py

@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
+# type: ignore
 
 from .main import MigrateOSDsHandler
 

src/rookify/modules/migrate_osds/main.py

@@ -2,15 +2,17 @@
 
 from ..module import ModuleHandler
 
+from typing import Any, Dict
+
 
 class MigrateOSDsHandler(ModuleHandler):
-    def preflight_check(self):
+    def preflight_check(self) -> None:
         pass
         # result = self.ceph.mon_command("osd dump")
         # raise ModuleException('test error')
 
-    def run(self) -> dict:
-        osd_config = dict()
+    def run(self) -> Dict[str, Any]:
+        osd_config: Dict[str, Any] = dict()
         for node, osds in self._data["analyze_ceph"]["node"]["ls"]["osd"].items():
             osd_config[node] = {"osds": {}}
             for osd in osds:
@@ -33,3 +35,4 @@ def run(self) -> dict:
                         break
 
         print(osd_config)
+        return {}

src/rookify/modules/module.py

@@ -5,6 +5,7 @@
 import rados
 import kubernetes
 import fabric
+from typing import Any, Dict, List, Optional
 
 
 class ModuleException(Exception):
@@ -17,7 +18,7 @@ class ModuleHandler:
     """
 
     class __Ceph:
-        def __init__(self, config: dict):
+        def __init__(self, config: Dict[str, Any]):
             try:
                 self.__ceph = rados.Rados(
                     conffile=config["conf_file"], conf={"keyring": config["keyring"]}
@@ -26,16 +27,20 @@ def __init__(self, config: dict):
             except rados.ObjectNotFound as err:
                 raise ModuleException(f"Could not connect to ceph: {err}")
 
-        def mon_command(self, command: str, **kwargs) -> dict:
+        def mon_command(
+            self, command: str, **kwargs: str
+        ) -> Dict[str, Any] | List[Any]:
             cmd = {"prefix": command, "format": "json"}
-            cmd.update(kwargs)
+            cmd.update(**kwargs)
             result = self.__ceph.mon_command(json.dumps(cmd), b"")
             if result[0] != 0:
                 raise ModuleException(f"Ceph did return an error: {result}")
-            return json.loads(result[1])
+            data = json.loads(result[1])
+            assert isinstance(data, dict) or isinstance(data, list)
+            return data
 
     class __K8s:
-        def __init__(self, config: dict):
+        def __init__(self, config: Dict[str, Any]):
             k8s_config = kubernetes.client.Configuration()
             k8s_config.api_key = config["api_key"]
             k8s_config.host = config["host"]
@@ -54,7 +59,7 @@ def NodeV1Api(self) -> kubernetes.client.NodeV1Api:
             return kubernetes.client.NodeV1Api(self.__client)
 
     class __SSH:
-        def __init__(self, config: dict):
+        def __init__(self, config: Dict[str, Any]):
             self.__config = config
 
         def command(self, host: str, command: str) -> fabric.runners.Result:
@@ -77,7 +82,7 @@ def command(self, host: str, command: str) -> fabric.runners.Result:
             ).run(command, hide=True)
             return result
 
-    def __init__(self, config: dict, data: dict):
+    def __init__(self, config: Dict[str, Any], data: Dict[str, Any]):
         """
         Construct a new 'ModuleHandler' object.
 
@@ -86,9 +91,9 @@ def __init__(self, config: dict, data: dict):
         """
         self._config = config
         self._data = data
-        self.__ceph = None
-        self.__k8s = None
-        self.__ssh = None
+        self.__ceph: Optional[ModuleHandler.__Ceph] = None
+        self.__k8s: Optional[ModuleHandler.__K8s] = None
+        self.__ssh: Optional[ModuleHandler.__SSH] = None
 
     @abc.abstractmethod
     def preflight_check(self) -> None:
@@ -98,7 +103,7 @@ def preflight_check(self) -> None:
         pass
 
     @abc.abstractmethod
-    def run(self) -> dict:
+    def run(self) -> Dict[str, Any]:
         """
         Run the modules tasks
 
@@ -109,17 +114,17 @@ def run(self) -> dict:
     @property
     def ceph(self) -> __Ceph:
         if self.__ceph is None:
-            self.__ceph = self.__Ceph(self._config["ceph"])
+            self.__ceph = ModuleHandler.__Ceph(self._config["ceph"])
         return self.__ceph
 
     @property
     def k8s(self) -> __K8s:
         if self.__k8s is None:
-            self.__k8s = self.__K8s(self._config["kubernetes"])
+            self.__k8s = ModuleHandler.__K8s(self._config["kubernetes"])
         return self.__k8s
 
     @property
     def ssh(self) -> __SSH:
         if self.__ssh is None:
-            self.__ssh = self.__SSH(self._config["ssh"])
+            self.__ssh = ModuleHandler.__SSH(self._config["ssh"])
         return self.__ssh

src/rookify/yaml.py

@@ -1,13 +1,16 @@
 # -*- coding: utf-8 -*-
 
 import yaml
+from typing import Any, Dict
 
 
-def load_yaml(path: str) -> dict:
+def load_yaml(path: str) -> Dict[str, Any]:
     with open(path, "r") as file:
-        return yaml.safe_load(file)
+        data = yaml.safe_load(file)
+        assert isinstance(data, dict)
+        return data
 
 
-def save_yaml(path: str, data: dict) -> None:
+def save_yaml(path: str, data: Dict[str, Any]) -> None:
     with open(path, "w") as file:
         yaml.safe_dump(data, file)