More file checking

Author: davidbrochart

nbclient/__init__.py

@@ -4,7 +4,7 @@
 from ._version import version as __version__  # noqa: F401
 
 
-def _cleanup():
+def _cleanup() -> None:
     pass
 
 

nbclient/client.py

@@ -992,7 +992,7 @@ def _get_buffer_data(self, msg: t.Dict) -> t.List[t.Dict[str, str]]:
     def register_output_hook(
             self,
             msg_id: str,
-            hook: t.Callable) -> None:
+            hook: OutputWidget) -> None:
         """Registers an override object that handles output/clear_output instead.
 
         Multiple hooks can be registered, where the last one will be used (stack based)
@@ -1004,7 +1004,7 @@ def register_output_hook(
     def remove_output_hook(
             self,
             msg_id: str,
-            hook: t.Callable) -> None:
+            hook: OutputWidget) -> None:
         """Unregisters an override object that handles output/clear_output instead"""
         # mimics
         # https://jupyterlab.github.io/jupyterlab/services/interfaces/kernel.ikernelconnection.html#removemessagehook

nbclient/exceptions.py

@@ -1,3 +1,8 @@
+from typing import Dict
+
+from nbformat import NotebookNode
+
+
 class CellControlSignal(Exception):
     """
     A custom exception used to indicate that the exception is used for cell
@@ -13,7 +18,11 @@ class CellTimeoutError(TimeoutError, CellControlSignal):
     """
 
     @classmethod
-    def error_from_timeout_and_cell(cls, msg, timeout, cell):
+    def error_from_timeout_and_cell(
+            cls,
+            msg: str,
+            timeout: int,
+            cell: NotebookNode) -> CellTimeoutError:
         if cell and cell.source:
             src_by_lines = cell.source.strip().split("\n")
             src = (
@@ -49,23 +58,30 @@ class CellExecutionError(CellControlSignal):
     failures gracefully.
     """
 
-    def __init__(self, traceback, ename, evalue):
+    def __init__(
+            self,
+            traceback: str,
+            ename: str,
+            evalue: str) -> None:
         super(CellExecutionError, self).__init__(traceback)
         self.traceback = traceback
         self.ename = ename
         self.evalue = evalue
 
-    def __str__(self):
+    def __str__(self) -> str:
         s = self.__unicode__()
         if not isinstance(s, str):
             s = s.encode('utf8', 'replace')
         return s
 
-    def __unicode__(self):
+    def __unicode__(self) -> str:
         return self.traceback
 
     @classmethod
-    def from_cell_and_msg(cls, cell, msg):
+    def from_cell_and_msg(
+            cls,
+            cell: NotebookNode,
+            msg: Dict) -> CellExecutionError:
         """Instantiate from a code cell object and a message contents
         (message is either execute_reply or error)
         """
@@ -82,7 +98,7 @@ def from_cell_and_msg(cls, cell, msg):
         )
 
 
-exec_err_msg = u"""\
+exec_err_msg: str = u"""\
 An error occurred while executing the following cell:
 ------------------
 {cell.source}
@@ -93,7 +109,7 @@ def from_cell_and_msg(cls, cell, msg):
 """
 
 
-timeout_err_msg = u"""\
+timeout_err_msg: str = u"""\
 A cell timed out while it was being executed, after {timeout} seconds.
 The message was: {msg}.
 Here is a preview of the cell contents:

nbclient/jsonutil.py

@@ -11,6 +11,7 @@
 import types
 from datetime import datetime
 import numbers
+from typing import Dict
 
 
 from ipython_genutils import py3compat
@@ -51,7 +52,7 @@
 PDF64 = b'JVBER'
 
 
-def encode_images(format_dict):
+def encode_images(format_dict: Dict) -> Dict[str, str]:
     """b64-encodes images in a displaypub format dict
 
     Perhaps this should be handled in json_clean itself?

nbclient/output_widget.py

@@ -1,19 +1,34 @@
 from .jsonutil import json_clean
 from nbformat.v4 import output_from_msg
+from typing import Dict, List, Any, Optional
+
+from jupyter_client.client import KernelClient
+from nbclient import NotebookClient
 
 
 class OutputWidget:
     """This class mimics a front end output widget"""
-    def __init__(self, comm_id, state, kernel_client, executor):
-        self.comm_id = comm_id
-        self.state = state
-        self.kernel_client = kernel_client
-        self.executor = executor
-        self.topic = ('comm-%s' % self.comm_id).encode('ascii')
-        self.outputs = self.state['outputs']
-        self.clear_before_next_output = False
+    def __init__(
+            self,
+            comm_id: str,
+            state: Dict[str, Any],
+            kernel_client: KernelClient,
+            executor: NotebookClient) -> None:
+
+        self.comm_id: str = comm_id
+        self.state: Dict[str, Any] = state
+        self.kernel_client: KernelClient = kernel_client
+        self.executor: NotebookClient = executor
+        self.topic: bytes = ('comm-%s' % self.comm_id).encode('ascii')
+        self.outputs: List = self.state['outputs']
+        self.clear_before_next_output: bool = False
+
+    def clear_output(
+            self,
+            outs: List,
+            msg: Dict,
+            cell_index: int) -> None:
 
-    def clear_output(self, outs, msg, cell_index):
         self.parent_header = msg['parent_header']
         content = msg['content']
         if content.get('wait'):
@@ -26,12 +41,18 @@ def clear_output(self, outs, msg, cell_index):
                 # sync the state to the nbconvert state as well, since that is used for testing
                 self.executor.widget_state[self.comm_id]['outputs'] = self.outputs
 
-    def sync_state(self):
+    def sync_state(self) -> None:
         state = {'outputs': self.outputs}
         msg = {'method': 'update', 'state': state, 'buffer_paths': []}
         self.send(msg)
 
-    def _publish_msg(self, msg_type, data=None, metadata=None, buffers=None, **keys):
+    def _publish_msg(
+            self,
+            msg_type: str,
+            data: Optional[Dict] = None,
+            metadata: Optional[Dict] = None,
+            buffers: Optional[List] = None,
+            **keys) -> None:
         """Helper for sending a comm message on IOPub"""
         data = {} if data is None else data
         metadata = {} if metadata is None else metadata
@@ -40,10 +61,21 @@ def _publish_msg(self, msg_type, data=None, metadata=None, buffers=None, **keys)
                                              metadata=metadata)
         self.kernel_client.shell_channel.send(msg)
 
-    def send(self, data=None, metadata=None, buffers=None):
+    def send(
+            self,
+            data: Optional[Dict] = None,
+            metadata: Optional[Dict] = None,
+            buffers: Optional[List] = None) -> None:
+
         self._publish_msg('comm_msg', data=data, metadata=metadata, buffers=buffers)
 
-    def output(self, outs, msg, display_id, cell_index):
+    def output(
+            self,
+            outs: List,
+            msg: Dict,
+            display_id: str,
+            cell_index: int) -> None:
+
         if self.clear_before_next_output:
             self.outputs = []
             self.clear_before_next_output = False
@@ -66,7 +98,7 @@ def output(self, outs, msg, display_id, cell_index):
             # sync the state to the nbconvert state as well, since that is used for testing
             self.executor.widget_state[self.comm_id]['outputs'] = self.outputs
 
-    def set_state(self, state):
+    def set_state(self, state: Dict) -> None:
         if 'msg_id' in state:
             msg_id = state.get('msg_id')
             if msg_id:
@@ -76,7 +108,7 @@ def set_state(self, state):
                 self.executor.remove_output_hook(self.msg_id, self)
                 self.msg_id = msg_id
 
-    def handle_msg(self, msg):
+    def handle_msg(self, msg: Dict) -> None:
         content = msg['content']
         comm_id = content['comm_id']
         assert comm_id == self.comm_id

tox.ini

@@ -18,7 +18,7 @@ commands = flake8 nbclient --count --ignore=E203,E731,F811,W503 --max-complexity
 [testenv:mypy]
 skip_install = true
 deps = mypy
-commands = mypy nbclient/client.py nbclient/util.py
+commands = mypy nbclient/*.py
 
 # Manifest
 [testenv:manifest]