Add annotations to nornir.plugins and mypy to CI (#187)

.gitignore

@@ -101,5 +101,6 @@ output/
 .DS_Store
 
 .pytest_cache/
+.mypy_cache/
 
 .vscode

.travis.yml

@@ -17,6 +17,8 @@ matrix:
     env: TOXENV=black
   - python: 3.6
     env: TOXENV=pylama
+  - python: 3.6
+    env: TOXENV=mypy
 install:
 - pip install tox tox-travis coveralls
 script:

nornir/core/inventory.py

@@ -1,5 +1,11 @@
+from typing import Dict, Any
+
 import getpass
 
+VarsDict = Dict[str, Any]
+HostsDict = Dict[str, VarsDict]
+GroupsDict = Dict[str, VarsDict]
+
 
 class Host(object):
     """

nornir/core/task.py

@@ -1,6 +1,7 @@
 import logging
 import traceback
-from builtins import super
+
+from typing import Optional
 
 from nornir.core.exceptions import NornirExecutionError
 from nornir.core.exceptions import NornirSubTaskError
@@ -163,6 +164,9 @@ def __init__(
         self.name = None
         self.severity_level = severity_level
 
+        self.stdout: Optional[str] = None
+        self.stderr: Optional[str] = None
+
         for k, v in kwargs.items():
             setattr(self, k, v)
 

nornir/plugins/functions/text/__init__.py

@@ -1,6 +1,7 @@
 import logging
 import pprint
 import threading
+from typing import List, Optional
 
 from colorama import Fore, Style, init
 
@@ -13,15 +14,15 @@
 init(autoreset=True, convert=False, strip=False)
 
 
-def print_title(title):
+def print_title(title: str) -> None:
     """
     Helper function to print a title.
     """
     msg = "**** {} ".format(title)
     print("{}{}{}{}".format(Style.BRIGHT, Fore.GREEN, msg, "*" * (80 - len(msg))))
 
 
-def _get_color(result, failed):
+def _get_color(result: Result, failed: bool) -> str:
     if result.failed or failed:
         color = Fore.RED
     elif result.changed:
@@ -32,8 +33,13 @@ def _get_color(result, failed):
 
 
 def _print_individual_result(
-    result, host, vars, failed, severity_level, task_group=False
-):
+    result: Result,
+    host: Optional[str],
+    attrs: List[str],
+    failed: bool,
+    severity_level: int,
+    task_group: bool = False,
+) -> None:
     if result.severity_level < severity_level:
         return
 
@@ -49,20 +55,24 @@ def _print_individual_result(
             Style.BRIGHT, color, msg, symbol * (80 - len(msg)), level_name
         )
     )
-    for v in vars:
-        x = getattr(result, v, "")
+    for attribute in attrs:
+        x = getattr(result, attribute, "")
         if x and not isinstance(x, str):
             pprint.pprint(x, indent=2)
         elif x:
             print(x)
 
 
 def _print_result(
-    result, host=None, vars=None, failed=None, severity_level=logging.INFO
-):
-    vars = vars or ["diff", "result", "stdout"]
-    if isinstance(vars, str):
-        vars = [vars]
+    result: Result,
+    host: Optional[str] = None,
+    attrs: List[str] = None,
+    failed: bool = False,
+    severity_level: int = logging.INFO,
+) -> None:
+    attrs = attrs or ["diff", "result", "stdout"]
+    if isinstance(attrs, str):
+        attrs = [attrs]
 
     if isinstance(result, AggregatedResult):
         msg = result.name
@@ -75,34 +85,36 @@ def _print_result(
             print(
                 "{}{}{}{}".format(Style.BRIGHT, Fore.BLUE, msg, "*" * (80 - len(msg)))
             )
-            _print_result(host_data, host, vars, failed, severity_level)
+            _print_result(host_data, host, attrs, failed, severity_level)
     elif isinstance(result, MultiResult):
         _print_individual_result(
-            result[0], host, vars, failed, severity_level, task_group=True
+            result[0], host, attrs, failed, severity_level, task_group=True
         )
         for r in result[1:]:
-            _print_result(r, host, vars, failed, severity_level)
+            _print_result(r, host, attrs, failed, severity_level)
         color = _get_color(result[0], failed)
         msg = "^^^^ END {} ".format(result[0].name)
         print("{}{}{}{}".format(Style.BRIGHT, color, msg, "^" * (80 - len(msg))))
     elif isinstance(result, Result):
-        _print_individual_result(result, host, vars, failed, severity_level)
+        _print_individual_result(result, host, attrs, failed, severity_level)
 
 
 def print_result(
-    result, host=None, vars=None, failed=None, severity_level=logging.INFO
-):
+    result: Result,
+    host: Optional[str] = None,
+    vars: List[str] = None,
+    failed: bool = False,
+    severity_level: int = logging.INFO,
+) -> None:
     """
     Prints the :obj:`nornir.core.task.Result` from a previous task to screen
 
     Arguments:
-        result (:obj:`nornir.core.task.Result`): from a previous task
-        vars (list of str): Which attributes you want to print
-        failed (``bool``): if ``True`` assume the task failed
-        severity_level (int): Print only errors with this severity level or higher
-
-    Returns:
-        :obj:`nornir.core.task.Result`:
+        result: from a previous task
+        host: # TODO
+        vars: Which attributes you want to print
+        failed: if ``True`` assume the task failed
+        severity_level: Print only errors with this severity level or higher
     """
     LOCK.acquire()
     try: