Add type annotations

.gitignore

@@ -39,6 +39,7 @@ htmlcov/
 .coverage
 .coverage.*
 .cache
+.mypy_cache
 nosetests.xml
 coverage.xml
 *,cover

.pre-commit-config.yaml

@@ -17,6 +17,13 @@ repos:
     -   id: trailing-whitespace
     -   id: end-of-file-fixer
     -   id: flake8
+        additional_dependencies: [flake8-typing-imports]
+-   repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v0.720
+    hooks:
+    -   id: mypy
+        files: ^(src/|testing/)
+        args: []
 -   repo: local
     hooks:
     -   id: rst

docs/index.rst

@@ -777,7 +777,9 @@ using the :py:meth:`pluggy._HookCaller.call_historic()` method:
 
 
     # call with history; no results returned
-    pm.hook.myhook.call_historic(config=config, args=sys.argv, result_callback=callback)
+    pm.hook.myhook.call_historic(
+        kwargs={"config": config, "args": sys.argv}, result_callback=callback
+    )
 
     # ... more of our program ...
 

setup.cfg

@@ -6,3 +6,20 @@ license_file=LICENSE
 
 [devpi:upload]
 formats=sdist.tgz,bdist_wheel
+
+[mypy]
+check_untyped_defs = True
+ignore_missing_imports = True
+no_implicit_optional = True
+strict_equality = True
+warn_redundant_casts = True
+warn_return_any = True
+warn_unused_configs = True
+warn_unused_ignores = True
+[mypy-pluggy.*]
+disallow_any_decorated = True
+disallow_any_generics = True
+disallow_any_unimported = True
+disallow_subclassing_any = True
+disallow_untyped_calls = True
+disallow_untyped_defs = True

src/pluggy/_tracing.py

@@ -3,17 +3,39 @@
 """
 from .callers import _Result
 
+if False:  # TYPE_CHECKING
+    from typing import Any
+    from typing import Callable
+    from typing import Dict
+    from typing import List
+    from typing import Optional
+    from typing import Sequence
+    from typing import Tuple
+    from typing import Union
+
+    from .hooks import HookImpl
+    from .hooks import _HookCaller
+    from .manager import PluginManager
+
+    _Writer = Callable[[str], None]
+    _Processor = Callable[[Tuple[str, ...], Sequence[object]], None]
+    _BeforeTrace = Callable[[str, List[HookImpl], Dict[str, Any]], None]
+    _AfterTrace = Callable[[_Result[Any], str, List[HookImpl], Dict[str, Any]], None]
+
 
 class TagTracer(object):
     def __init__(self):
-        self._tag2proc = {}
-        self.writer = None
+        # type: () -> None
+        self._tag2proc = {}  # type: Dict[Tuple[str, ...], _Processor]
+        self.writer = None  # type: Optional[_Writer]
         self.indent = 0
 
     def get(self, name):
+        # type: (str) -> TagTracerSub
         return TagTracerSub(self, (name,))
 
     def format_message(self, tags, args):
+        # type: (Sequence[str], Sequence[object]) -> List[str]
         if isinstance(args[-1], dict):
             extra = args[-1]
             args = args[:-1]
@@ -30,6 +52,7 @@ def format_message(self, tags, args):
         return lines
 
     def processmessage(self, tags, args):
+        # type: (Tuple[str, ...], Sequence[object]) -> None
         if self.writer is not None and args:
             lines = self.format_message(tags, args)
             self.writer("".join(lines))
@@ -39,9 +62,11 @@ def processmessage(self, tags, args):
             pass
 
     def setwriter(self, writer):
+        # type: (_Writer) -> None
         self.writer = writer
 
     def setprocessor(self, tags, processor):
+        # type: (Union[str, Tuple[str, ...]], _Processor) -> None
         if isinstance(tags, str):
             tags = tuple(tags.split(":"))
         else:
@@ -51,33 +76,40 @@ def setprocessor(self, tags, processor):
 
 class TagTracerSub(object):
     def __init__(self, root, tags):
+        # type: (TagTracer, Tuple[str, ...]) -> None
         self.root = root
         self.tags = tags
 
     def __call__(self, *args):
+        # type: (object) -> None
         self.root.processmessage(self.tags, args)
 
     def setmyprocessor(self, processor):
+        # type: (_Processor) -> None
         self.root.setprocessor(self.tags, processor)
 
     def get(self, name):
+        # type: (str) -> TagTracerSub
         return self.__class__(self.root, self.tags + (name,))
 
 
 class _TracedHookExecution(object):
     def __init__(self, pluginmanager, before, after):
+        # type: (PluginManager, _BeforeTrace, _AfterTrace) -> None
         self.pluginmanager = pluginmanager
         self.before = before
         self.after = after
         self.oldcall = pluginmanager._inner_hookexec
         assert not isinstance(self.oldcall, _TracedHookExecution)
         self.pluginmanager._inner_hookexec = self
 
-    def __call__(self, hook, hook_impls, kwargs):
-        self.before(hook.name, hook_impls, kwargs)
-        outcome = _Result.from_call(lambda: self.oldcall(hook, hook_impls, kwargs))
-        self.after(outcome, hook.name, hook_impls, kwargs)
+    def __call__(self, hook, methods, kwargs):
+        # type: (_HookCaller, List[HookImpl], Dict[str, object]) -> Union[object, List[object]]
+        self.before(hook.name, methods, kwargs)
+        outcome = _Result.from_call(lambda: self.oldcall(hook, methods, kwargs))
+        self.after(outcome, hook.name, methods, kwargs)
         return outcome.get_result()
 
     def undo(self):
+        # type: () -> None
         self.pluginmanager._inner_hookexec = self.oldcall

src/pluggy/callers.py

@@ -15,8 +15,51 @@ def _reraise(cls, val, tb):
 """
     )
 
+if False:  # TYPE_CHECKING
+    from types import TracebackType
+    from typing import Callable
+    from typing import cast
+    from typing import Dict
+    from typing import Generator
+    from typing import Generic
+    from typing import List
+    from typing import NoReturn
+    from typing import Optional
+    from typing import Tuple
+    from typing import Type
+    from typing import TypeVar
+    from typing import Union
+
+    from .hooks import HookImpl
+
+    _T = TypeVar("_T")
+
+    _ExcInfo = Tuple[Type[BaseException], BaseException, TracebackType]
+
+    _WrapController = Generator[None, "_Result[_T]", None]
+    _HookImplFunction = Callable[..., Union[_T, _WrapController[_T]]]
+
+    def _reraise(cls, val, tb):
+        # type: (Type[BaseException], BaseException, TracebackType) -> NoReturn
+        pass
+
+
+else:
+
+    def cast(x, y):  # type: ignore
+        return y
+
+    _T = object()  # type: ignore
+
+    class _GenericMeta:
+        def __getitem__(self, parameter):  # type: ignore
+            return object
+
+    Generic = _GenericMeta()  # type: ignore
+
 
 def _raise_wrapfail(wrap_controller, msg):
+    # type: (_WrapController[_T], str) -> NoReturn
     co = wrap_controller.gi_code
     raise RuntimeError(
         "wrap_controller at %r %s:%d %s"
@@ -28,34 +71,43 @@ class HookCallError(Exception):
     """ Hook was called wrongly. """
 
 
-class _Result(object):
+class _Result(Generic[_T]):
     def __init__(self, result, excinfo):
+        # type: (Optional[_T], Optional[_ExcInfo]) -> None
         self._result = result
         self._excinfo = excinfo
 
     @property
     def excinfo(self):
+        # type: () -> Optional[_ExcInfo]
         return self._excinfo
 
     @property
     def result(self):
+        # type: () -> Optional[_T]
         """Get the result(s) for this hook call (DEPRECATED in favor of ``get_result()``)."""
         msg = "Use get_result() which forces correct exception handling"
         warnings.warn(DeprecationWarning(msg), stacklevel=2)
         return self._result
 
     @classmethod
     def from_call(cls, func):
+        # type: (Callable[[], _T]) -> _Result[_T]
         __tracebackhide__ = True
         result = excinfo = None
         try:
             result = func()
         except BaseException:
-            excinfo = sys.exc_info()
+            _excinfo = sys.exc_info()
+            assert _excinfo[0] is not None
+            assert _excinfo[1] is not None
+            assert _excinfo[2] is not None
+            excinfo = (_excinfo[0], _excinfo[1], _excinfo[2])
 
         return cls(result, excinfo)
 
     def force_result(self, result):
+        # type: (_T) -> None
         """Force the result(s) to ``result``.
 
         If the hook was marked as a ``firstresult`` a single value should
@@ -66,14 +118,15 @@ def force_result(self, result):
         self._excinfo = None
 
     def get_result(self):
+        # type: () -> _T
         """Get the result(s) for this hook call.
 
         If the hook was marked as a ``firstresult`` only a single value
         will be returned otherwise a list of results.
         """
         __tracebackhide__ = True
         if self._excinfo is None:
-            return self._result
+            return self._result  # type: ignore
         else:
             ex = self._excinfo
             if _py3:
@@ -82,6 +135,7 @@ def get_result(self):
 
 
 def _wrapped_call(wrap_controller, func):
+    # type: (_WrapController[_T], Callable[[], _T]) -> _T
     """ Wrap calling to a function with a generator which needs to yield
     exactly once.  The yield point will trigger calling the wrapped function
     and return its ``_Result`` to the yield point.  The generator then needs
@@ -110,14 +164,17 @@ class _LegacyMultiCall(object):
     # in execute() and simplify/speed up the execute loop.
 
     def __init__(self, hook_impls, kwargs, firstresult=False):
+        # type: (List[HookImpl], Dict[str, object], bool) -> None
         self.hook_impls = hook_impls
         self.caller_kwargs = kwargs  # come from _HookCaller.__call__()
         self.caller_kwargs["__multicall__"] = self
         self.firstresult = firstresult
 
     def execute(self):
+        # type: () -> Union[object, Optional[List[object]]]
         caller_kwargs = self.caller_kwargs
-        self.results = results = []
+        results = []  # type: List[object]
+        self.results = results
         firstresult = self.firstresult
 
         while self.hook_impls:
@@ -130,9 +187,12 @@ def execute(self):
                         raise HookCallError(
                             "hook call must provide argument %r" % (argname,)
                         )
-            if hook_impl.hookwrapper:
-                return _wrapped_call(hook_impl.function(*args), self.execute)
             res = hook_impl.function(*args)
+            if hook_impl.hookwrapper:
+                # If this cast is not valid, a type error is raised below,
+                # which is the desired response.
+                gen = cast("_WrapController[object]", res)
+                return _wrapped_call(gen, self.execute)
             if res is not None:
                 if firstresult:
                     return res
@@ -141,27 +201,32 @@ def execute(self):
         if not firstresult:
             return results
 
+        return None
+
     def __repr__(self):
+        # type: () -> str
         status = "%d meths" % (len(self.hook_impls),)
         if hasattr(self, "results"):
             status = ("%d results, " % len(self.results)) + status
         return "<_MultiCall %s, kwargs=%r>" % (status, self.caller_kwargs)
 
 
 def _legacymulticall(hook_impls, caller_kwargs, firstresult=False):
+    # type: (List[HookImpl], Dict[str, object], bool) -> Union[object, Optional[List[object]]]
     return _LegacyMultiCall(
         hook_impls, caller_kwargs, firstresult=firstresult
     ).execute()
 
 
 def _multicall(hook_impls, caller_kwargs, firstresult=False):
+    # type: (List[HookImpl], Dict[str, object], bool) -> Union[object, List[object]]
     """Execute a call into multiple python functions/methods and return the
     result(s).
 
     ``caller_kwargs`` comes from _HookCaller.__call__().
     """
     __tracebackhide__ = True
-    results = []
+    results = []  # type: List[object]
     excinfo = None
     try:  # run impl and wrapper setup functions in a loop
         teardowns = []
@@ -176,24 +241,32 @@ def _multicall(hook_impls, caller_kwargs, firstresult=False):
                                 "hook call must provide argument %r" % (argname,)
                             )
 
+                res = hook_impl.function(*args)
                 if hook_impl.hookwrapper:
+                    # If this cast is not valid, a type error is raised below,
+                    # which is the desired response.
+                    gen = cast("_WrapController[object]", res)
                     try:
-                        gen = hook_impl.function(*args)
                         next(gen)  # first yield
                         teardowns.append(gen)
                     except StopIteration:
                         _raise_wrapfail(gen, "did not yield")
                 else:
-                    res = hook_impl.function(*args)
                     if res is not None:
                         results.append(res)
                         if firstresult:  # halt further impl calls
                             break
         except BaseException:
-            excinfo = sys.exc_info()
+            _excinfo = sys.exc_info()
+            assert _excinfo[0] is not None
+            assert _excinfo[1] is not None
+            assert _excinfo[2] is not None
+            excinfo = (_excinfo[0], _excinfo[1], _excinfo[2])
     finally:
         if firstresult:  # first result hooks return a single value
-            outcome = _Result(results[0] if results else None, excinfo)
+            outcome = _Result(
+                results[0] if results else None, excinfo
+            )  # type: _Result[Union[object, List[object]]]
         else:
             outcome = _Result(results, excinfo)