Use jupyter_client's AsyncKernelManager

docs/source/extending/bundler_extensions.rst

@@ -86,20 +86,19 @@ respond in any manner. For example, it may read additional query parameters
 from the request, issue a redirect to another site, run a local process (e.g.,
 `nbconvert`), make a HTTP request to another service, etc.
 
-The caller of the `bundle` function is `@tornado.gen.coroutine` decorated and
-wraps its call with `torando.gen.maybe_future`. This behavior means you may
+The caller of the `bundle` function is `async` and wraps its call with
+`jupyter_server.utils.ensure_async`. This behavior means you may
 handle the web request synchronously, as in the example above, or
-asynchronously using `@tornado.gen.coroutine` and `yield`, as in the example
+asynchronously using `async` and `await`, as in the example
 below.
 
 .. code:: python
 
-    from tornado import gen
+    import asyncio
 
-    @gen.coroutine
-    def bundle(handler, model):
+    async def bundle(handler, model):
       # simulate a long running IO op (e.g., deploying to a remote host)
-      yield gen.sleep(10)
+      await asyncio.sleep(10)
 
       # now respond
       handler.finish('I spent 10 seconds bundling {}!'.format(model['path']))

jupyter_server/base/zmqhandlers.py

@@ -10,15 +10,14 @@
 import tornado
 
 from urllib.parse import urlparse
-from tornado import gen, ioloop, web
+from tornado import ioloop, web
 from tornado.websocket import WebSocketHandler
 
 from jupyter_client.session import Session
 from jupyter_client.jsonutil import date_default, extract_dates
 from ipython_genutils.py3compat import cast_unicode
 
 from .handlers import JupyterHandler
-from jupyter_server.utils import maybe_future
 
 
 def serialize_binary_message(msg):
@@ -90,15 +89,15 @@ class WebSocketMixin(object):
     last_ping = 0
     last_pong = 0
     stream = None
-    
+
     @property
     def ping_interval(self):
         """The interval for websocket keep-alive pings.
-        
+
         Set ws_ping_interval = 0 to disable pings.
         """
         return self.settings.get('ws_ping_interval', WS_PING_INTERVAL)
-    
+
     @property
     def ping_timeout(self):
         """If no ping is received in this many milliseconds,
@@ -111,7 +110,7 @@ def ping_timeout(self):
 
     def check_origin(self, origin=None):
         """Check Origin == Host or Access-Control-Allow-Origin.
-        
+
         Tornado >= 4 calls this method automatically, raising 403 if it returns False.
         """
 
@@ -122,18 +121,18 @@ def check_origin(self, origin=None):
         host = self.request.headers.get("Host")
         if origin is None:
             origin = self.get_origin()
-        
+
         # If no origin or host header is provided, assume from script
         if origin is None or host is None:
             return True
-        
+
         origin = origin.lower()
         origin_host = urlparse(origin).netloc
-        
+
         # OK if origin matches host
         if origin_host == host:
             return True
-        
+
         # Check CORS headers
         if self.allow_origin:
             allow = self.allow_origin == origin
@@ -190,7 +189,7 @@ def on_pong(self, data):
 
 
 class ZMQStreamHandler(WebSocketMixin, WebSocketHandler):
-    
+
     if tornado.version_info < (4,1):
         """Backport send_error from tornado 4.1 to 4.0"""
         def send_error(self, *args, **kwargs):
@@ -203,17 +202,17 @@ def send_error(self, *args, **kwargs):
                 # we can close the connection more gracefully.
                 self.stream.close()
 
-    
+
     def _reserialize_reply(self, msg_or_list, channel=None):
         """Reserialize a reply message using JSON.
 
         msg_or_list can be an already-deserialized msg dict or the zmq buffer list.
         If it is the zmq list, it will be deserialized with self.session.
-        
+
         This takes the msg list from the ZMQ socket and serializes the result for the websocket.
         This method should be used by self._on_zmq_reply to build messages that can
         be sent back to the browser.
-        
+
         """
         if isinstance(msg_or_list, dict):
             # already unpacked
@@ -271,14 +270,13 @@ def pre_get(self):
         else:
             self.log.warning("No session ID specified")
 
-    @gen.coroutine
-    def get(self, *args, **kwargs):
+    async def get(self, *args, **kwargs):
         # pre_get can be a coroutine in subclasses
         # assign and yield in two step to avoid tornado 3 issues
         res = self.pre_get()
-        yield maybe_future(res)
+        await res
         res = super(AuthenticatedZMQStreamHandler, self).get(*args, **kwargs)
-        yield maybe_future(res)
+        await res
 
     def initialize(self):
         self.log.debug("Initializing websocket connection %s", self.request.path)

jupyter_server/files/handlers.py

@@ -6,9 +6,8 @@
 import mimetypes
 import json
 from base64 import decodebytes
-from tornado import gen, web
+from tornado import web
 from jupyter_server.base.handlers import JupyterHandler
-from jupyter_server.utils import maybe_future
 
 
 class FilesHandler(JupyterHandler):
@@ -32,8 +31,7 @@ def head(self, path):
         self.get(path, include_body=False)
 
     @web.authenticated
-    @gen.coroutine
-    def get(self, path, include_body=True):
+    async def get(self, path, include_body=True):
         cm = self.contents_manager
 
         if cm.is_hidden(path) and not cm.allow_hidden:
@@ -45,9 +43,9 @@ def get(self, path, include_body=True):
             _, name = path.rsplit('/', 1)
         else:
             name = path
-        
-        model = yield maybe_future(cm.get(path, type='file', content=include_body))
-        
+
+        model = await cm.get(path, type='file', content=include_body)
+
         if self.get_argument("download", False):
             self.set_attachment_header(name)
 
@@ -78,4 +76,3 @@ def get(self, path, include_body=True):
 
 
 default_handlers = []
-

jupyter_server/gateway/handlers.py

@@ -8,7 +8,7 @@
 from ..base.handlers import APIHandler, JupyterHandler
 from ..utils import url_path_join
 
-from tornado import gen, web
+from tornado import web
 from tornado.concurrent import Future
 from tornado.ioloop import IOLoop, PeriodicCallback
 from tornado.websocket import WebSocketHandler, websocket_connect
@@ -61,11 +61,10 @@ def initialize(self):
         self.session = Session(config=self.config)
         self.gateway = GatewayWebSocketClient(gateway_url=GatewayClient.instance().url)
 
-    @gen.coroutine
-    def get(self, kernel_id, *args, **kwargs):
+    async def get(self, kernel_id, *args, **kwargs):
         self.authenticate()
         self.kernel_id = cast_unicode(kernel_id, 'ascii')
-        yield super(WebSocketChannelsHandler, self).get(kernel_id=kernel_id, *args, **kwargs)
+        await super(WebSocketChannelsHandler, self).get(kernel_id=kernel_id, *args, **kwargs)
 
     def send_ping(self):
         if self.ws_connection is None and self.ping_callback is not None:
@@ -132,8 +131,7 @@ def __init__(self, **kwargs):
         self.ws_future = Future()
         self.disconnected = False
 
-    @gen.coroutine
-    def _connect(self, kernel_id):
+    async def _connect(self, kernel_id):
         # websocket is initialized before connection
         self.ws = None
         self.kernel_id = kernel_id
@@ -168,14 +166,13 @@ def _disconnect(self):
             self.ws_future.cancel()
             self.log.debug("_disconnect: future cancelled, disconnected: {}".format(self.disconnected))
 
-    @gen.coroutine
-    def _read_messages(self, callback):
+    async def _read_messages(self, callback):
         """Read messages from gateway server."""
         while self.ws is not None:
             message = None
             if not self.disconnected:
                 try:
-                    message = yield self.ws.read_message()
+                    message = await self.ws.read_message()
                 except Exception as e:
                     self.log.error("Exception reading message from websocket: {}".format(e))  # , exc_info=True)
                 if message is None:
@@ -229,10 +226,9 @@ class GatewayResourceHandler(APIHandler):
     """Retrieves resources for specific kernelspec definitions from kernel/enterprise gateway."""
 
     @web.authenticated
-    @gen.coroutine
-    def get(self, kernel_name, path, include_body=True):
+    async def get(self, kernel_name, path, include_body=True):
         ksm = self.kernel_spec_manager
-        kernel_spec_res = yield ksm.get_kernel_spec_resource(kernel_name, path)
+        kernel_spec_res = await ksm.get_kernel_spec_resource(kernel_name, path)
         if kernel_spec_res is None:
             self.log.warning("Kernelspec resource '{}' for '{}' not found.  Gateway may not support"
                              " resource serving.".format(path, kernel_name))