Type check nbclient/util.py

Author: davidbrochart

nbclient/util.py

@@ -6,28 +6,29 @@
 import asyncio
 import sys
 import inspect
+from typing import Callable, Awaitable, Any, Union
 
 
-def check_ipython():
+def check_ipython() -> None:
     # original from vaex/asyncio.py
     IPython = sys.modules.get('IPython')
     if IPython:
-        IPython_version = tuple(map(int, IPython.__version__.split('.')))
+        IPython_version = tuple(map(int, IPython.__version__.split('.')))  # type: ignore
         if IPython_version < (7, 0, 0):
-            raise RuntimeError(f'You are using IPython {IPython.__version__} while we require'
+            raise RuntimeError(f'You are using IPython {IPython.__version__} while we require'  # type: ignore
                                '7.0.0+, please update IPython')
 
 
-def check_patch_tornado():
+def check_patch_tornado() -> None:
     """If tornado is imported, add the patched asyncio.Future to its tuple of acceptable Futures"""
     # original from vaex/asyncio.py
     if 'tornado' in sys.modules:
         import tornado.concurrent
         if asyncio.Future not in tornado.concurrent.FUTURES:
-            tornado.concurrent.FUTURES = tornado.concurrent.FUTURES + (asyncio.Future, )
+            tornado.concurrent.FUTURES = tornado.concurrent.FUTURES + (asyncio.Future, )  # type: ignore
 
 
-def just_run(coro):
+def just_run(coro: Awaitable) -> Any:
     """Make the coroutine run, even if there is an event loop running (using nest_asyncio)"""
     # original from vaex/asyncio.py
     loop = asyncio._get_running_loop()
@@ -51,7 +52,7 @@ def just_run(coro):
     return loop.run_until_complete(coro)
 
 
-def run_sync(coro):
+def run_sync(coro: Callable) -> Callable:
     """Runs a coroutine and blocks until it has executed.
 
     An event loop is created if no one already exists. If an event loop is
@@ -74,7 +75,7 @@ def wrapped(*args, **kwargs):
     return wrapped
 
 
-async def ensure_async(obj):
+async def ensure_async(obj: Union[Awaitable, Any]) -> Any:
     """Convert a non-awaitable object to a coroutine if needed,
     and await it if it was not already awaited.
     """

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
+commands = mypy nbclient/client.py nbclient/util.py
 
 # Manifest
 [testenv:manifest]