Strict type checking.

Author: jonathanslenders

.github/workflows/test.yaml

@@ -22,21 +22,21 @@ jobs:
         run: |
           sudo apt remove python3-pip
           python -m pip install --upgrade pip
-          python -m pip install . black coverage codecov flake8 isort==5.6.4 mypy pytest readme_renderer types-contextvars
+          python -m pip install . black coverage codecov flake8 isort==5.6.4 mypy pytest readme_renderer types-contextvars asyncssh
           pip list
       - name: Run Tests
         run: |
           flake8 prompt_toolkit
           coverage run -m pytest
       - name: Type Checker
         # Check wheather the imports were sorted correctly.
-        # When this fails, please run ./tools/sort-imports.sh      
+        # When this fails, please run ./tools/sort-imports.sh
         run: |
-          mypy prompt_toolkit
+          mypy --strict prompt_toolkit
           isort -c --profile black prompt_toolkit examples tests setup.py
           black --check prompt_toolkit examples tests setup.py
       - name: Validate README.md
-        # Ensure that the README renders correctly (required for uploading to PyPI).        
+        # Ensure that the README renders correctly (required for uploading to PyPI).
         run: |
           python -m readme_renderer README.rst > /dev/null
       - name: Run codecov

mypy.ini

@@ -1,6 +1,20 @@
 [mypy]
+platform = win32
+
+# --strict.
+check_untyped_defs = True
+disallow_any_generics = True
+disallow_incomplete_defs = True
+disallow_subclassing_any = True
+disallow_untyped_calls = True
+disallow_untyped_decorators = True
+disallow_untyped_defs = True
 ignore_missing_imports = True
 no_implicit_optional = True
-platform = win32
+no_implicit_reexport = True
 strict_equality = True
 strict_optional = True
+warn_redundant_casts = True
+warn_return_any = True
+warn_unused_configs = True
+warn_unused_ignores = True

prompt_toolkit/application/application.py

@@ -216,10 +216,10 @@ def __init__(
         max_render_postpone_time: Union[float, int, None] = 0.01,
         refresh_interval: Optional[float] = None,
         terminal_size_polling_interval: Optional[float] = 0.5,
-        on_reset: Optional[ApplicationEventHandler] = None,
-        on_invalidate: Optional[ApplicationEventHandler] = None,
-        before_render: Optional[ApplicationEventHandler] = None,
-        after_render: Optional[ApplicationEventHandler] = None,
+        on_reset: Optional["ApplicationEventHandler[_AppResult]"] = None,
+        on_invalidate: Optional["ApplicationEventHandler[_AppResult]"] = None,
+        before_render: Optional["ApplicationEventHandler[_AppResult]"] = None,
+        after_render: Optional["ApplicationEventHandler[_AppResult]"] = None,
         # I/O.
         input: Optional[Input] = None,
         output: Optional[Output] = None,
@@ -769,7 +769,7 @@ def flush_input() -> None:
                         # Store unprocessed input as typeahead for next time.
                         store_typeahead(self.input, self.key_processor.empty_queue())
 
-                return result
+                return cast(_AppResult, result)
 
         async def _run_async2() -> _AppResult:
             self._is_running = True

prompt_toolkit/application/run_in_terminal.py

@@ -1,16 +1,17 @@
 """
 Tools for running functions on the terminal above the current application or prompt.
 """
+import sys
 from asyncio import Future, ensure_future
 from typing import AsyncGenerator, Awaitable, Callable, TypeVar
 
 from prompt_toolkit.eventloop import run_in_executor_with_context
 
 from .current import get_app_or_none
 
-try:
-    from contextlib import asynccontextmanager  # type: ignore
-except ImportError:
+if sys.version_info >= (3, 7):
+    from contextlib import asynccontextmanager
+else:
     from prompt_toolkit.eventloop.async_context_manager import asynccontextmanager
 
 

prompt_toolkit/cache.py

@@ -101,7 +101,7 @@ def __missing__(self, key: _K) -> _V:
         return result
 
 
-_F = TypeVar("_F", bound=Callable)
+_F = TypeVar("_F", bound=Callable[..., object])
 
 
 def memoized(maxsize: int = 1024) -> Callable[[_F], _F]:

prompt_toolkit/contrib/ssh/server.py

@@ -3,7 +3,7 @@
 """
 import asyncio
 import traceback
-from typing import Awaitable, Callable, Optional, TextIO, cast
+from typing import Any, Awaitable, Callable, Optional, TextIO, cast
 
 import asyncssh
 
@@ -15,25 +15,25 @@
 __all__ = ["PromptToolkitSSHSession", "PromptToolkitSSHServer"]
 
 
-class PromptToolkitSSHSession(asyncssh.SSHServerSession):
+class PromptToolkitSSHSession(asyncssh.SSHServerSession):  # type: ignore
     def __init__(
         self, interact: Callable[["PromptToolkitSSHSession"], Awaitable[None]]
     ) -> None:
         self.interact = interact
         self.interact_task: Optional[asyncio.Task[None]] = None
-        self._chan = None
+        self._chan: Optional[Any] = None
         self.app_session: Optional[AppSession] = None
 
         # PipInput object, for sending input in the CLI.
         # (This is something that we can use in the prompt_toolkit event loop,
         # but still write date in manually.)
         self._input = create_pipe_input()
-        self._output = None
+        self._output: Optional[Vt100_Output] = None
 
         # Output object. Don't render to the real stdout, but write everything
         # in the SSH channel.
         class Stdout:
-            def write(s, data):
+            def write(s, data: str) -> None:
                 try:
                     if self._chan is not None:
                         self._chan.write(data.replace("\n", "\r\n"))
@@ -43,12 +43,13 @@ def write(s, data):
             def isatty(s) -> bool:
                 return True
 
-            def flush(s):
+            def flush(s) -> None:
                 pass
 
             @property
-            def encoding(s):
-                return self._chan._orig_chan.get_encoding()[0]
+            def encoding(s) -> str:
+                assert self._chan is not None
+                return str(self._chan._orig_chan.get_encoding()[0])
 
         self.stdout = cast(TextIO, Stdout())
 
@@ -62,7 +63,7 @@ def _get_size(self) -> Size:
             width, height, pixwidth, pixheight = self._chan.get_terminal_size()
             return Size(rows=height, columns=width)
 
-    def connection_made(self, chan):
+    def connection_made(self, chan: Any) -> None:
         self._chan = chan
 
     def shell_requested(self) -> bool:
@@ -98,17 +99,17 @@ async def _interact(self) -> None:
                 self._input.close()
 
     def terminal_size_changed(
-        self, width: int, height: int, pixwidth, pixheight
+        self, width: int, height: int, pixwidth: object, pixheight: object
     ) -> None:
         # Send resize event to the current application.
         if self.app_session and self.app_session.app:
             self.app_session.app._on_resize()
 
-    def data_received(self, data, datatype):
+    def data_received(self, data: str, datatype: object) -> None:
         self._input.send_text(data)
 
 
-class PromptToolkitSSHServer(asyncssh.SSHServer):
+class PromptToolkitSSHServer(asyncssh.SSHServer):  # type: ignore
     """
     Run a prompt_toolkit application over an asyncssh server.
 

prompt_toolkit/eventloop/async_context_manager.py

@@ -1,9 +1,12 @@
 """
 @asynccontextmanager code, copied from Python 3.7's contextlib.
 For usage in Python 3.6.
+Types have been added to this file, just enough to make Mypy happy.
 """
+# mypy: allow-untyped-defs
 import abc
 from functools import wraps
+from typing import TYPE_CHECKING, AsyncContextManager, AsyncIterator, Callable, TypeVar
 
 import _collections_abc
 
@@ -26,7 +29,7 @@ async def __aexit__(self, exc_type, exc_value, traceback):
     @classmethod
     def __subclasshook__(cls, C):
         if cls is AbstractAsyncContextManager:
-            return _collections_abc._check_methods(C, "__aenter__", "__aexit__")
+            return _collections_abc._check_methods(C, "__aenter__", "__aexit__")  # type: ignore
         return NotImplemented
 
 
@@ -95,7 +98,12 @@ async def __aexit__(self, typ, value, traceback):
                     raise
 
 
-def asynccontextmanager(func):
+_T = TypeVar("_T")
+
+
+def asynccontextmanager(
+    func: Callable[..., AsyncIterator[_T]]
+) -> Callable[..., AsyncContextManager[_T]]:
     """@asynccontextmanager decorator.
     Typical usage:
         @asynccontextmanager
@@ -119,6 +127,6 @@ async def some_async_generator(<arguments>):
 
     @wraps(func)
     def helper(*args, **kwds):
-        return _AsyncGeneratorContextManager(func, args, kwds)
+        return _AsyncGeneratorContextManager(func, args, kwds)  # type: ignore
 
     return helper

prompt_toolkit/eventloop/inputhook.py

@@ -28,8 +28,17 @@
 import selectors
 import threading
 from asyncio import AbstractEventLoop, get_event_loop
-from selectors import BaseSelector
-from typing import Callable
+from selectors import BaseSelector, SelectorKey
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    List,
+    Mapping,
+    NamedTuple,
+    Optional,
+    Tuple,
+)
 
 from prompt_toolkit.utils import is_windows
 
@@ -40,6 +49,11 @@
     "InputHookContext",
 ]
 
+if TYPE_CHECKING:
+    from _typeshed import FileDescriptor, FileDescriptorLike
+
+    _EventMask = int
+
 
 def new_eventloop_with_inputhook(
     inputhook: Callable[["InputHookContext"], None]
@@ -79,19 +93,25 @@ def __init__(
         self.inputhook = inputhook
         self._r, self._w = os.pipe()
 
-    def register(self, fileobj, events, data=None):
+    def register(
+        self, fileobj: "FileDescriptorLike", events: "_EventMask", data: Any = None
+    ) -> "SelectorKey":
         return self.selector.register(fileobj, events, data=data)
 
-    def unregister(self, fileobj):
+    def unregister(self, fileobj: "FileDescriptorLike") -> "SelectorKey":
         return self.selector.unregister(fileobj)
 
-    def modify(self, fileobj, events, data=None):
+    def modify(
+        self, fileobj: "FileDescriptorLike", events: "_EventMask", data: Any = None
+    ) -> "SelectorKey":
         return self.selector.modify(fileobj, events, data=None)
 
-    def select(self, timeout=None):
+    def select(
+        self, timeout: Optional[float] = None
+    ) -> List[Tuple["SelectorKey", "_EventMask"]]:
         # If there are tasks in the current event loop,
         # don't run the input hook.
-        if len(get_event_loop()._ready) > 0:
+        if len(getattr(get_event_loop(), "_ready", [])) > 0:
             return self.selector.select(timeout=timeout)
 
         ready = False
@@ -140,6 +160,7 @@ def input_is_ready() -> bool:
 
         # Wait for the real selector to be done.
         th.join()
+        assert result is not None
         return result
 
     def close(self) -> None:
@@ -153,7 +174,7 @@ def close(self) -> None:
         self._r = self._w = -1
         self.selector.close()
 
-    def get_map(self):
+    def get_map(self) -> Mapping["FileDescriptorLike", "SelectorKey"]:
         return self.selector.get_map()
 
 

prompt_toolkit/eventloop/win32.py

@@ -44,7 +44,7 @@ def wait_for_handles(
     arrtype = HANDLE * len(handles)
     handle_array = arrtype(*handles)
 
-    ret = windll.kernel32.WaitForMultipleObjects(
+    ret: int = windll.kernel32.WaitForMultipleObjects(
         len(handle_array), handle_array, BOOL(False), DWORD(timeout)
     )
 

prompt_toolkit/formatted_text/base.py

@@ -26,6 +26,7 @@
 
 
 if TYPE_CHECKING:
+    from typing_extensions import TypeGuard
 
     class MagicFormattedText(Protocol):
         """
@@ -100,7 +101,7 @@ def to_formatted_text(
         return FormattedText(result)
 
 
-def is_formatted_text(value: object) -> bool:
+def is_formatted_text(value: object) -> "TypeGuard[AnyFormattedText]":
     """
     Check whether the input is valid formatted text (for use in assert
     statements).