remove some use of old deprecated coroutine syntax

we can assume async def now

Author: minrk

ipyparallel/client/magics.py

@@ -43,17 +43,6 @@ def nullcontext():
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 # -----------------------------------------------------------------------------
-# -----------------------------------------------------------------------------
-# Imports
-# -----------------------------------------------------------------------------
-
-try:
-    from asyncio import coroutine
-except ImportError:  # py2
-
-    def coroutine(f):
-        raise NotImplementedError()
-
 
 import inspect
 import re
@@ -79,6 +68,20 @@ def _iscoroutinefunction(f):
     return False
 
 
+def _asyncify(f):
+    """Wrap a blocking call in a coroutine
+
+    Does not make the call non-blocking,
+    just conforms to the API assuming it is awaitable.
+    For use when patching-in replacement methods.
+    """
+
+    async def async_f(*args, **kwargs):
+        return f(*args, **kwargs)
+
+    return async_f
+
+
 NO_LAST_RESULT = "%pxresult recalls last %px result, which has not yet been used."
 
 
@@ -444,14 +447,14 @@ def _enable_autopx(self):
         if _iscoroutinefunction(self.shell.run_cell):
             # original is a coroutine,
             # wrap ours in a coroutine
-            pxrun_cell = coroutine(pxrun_cell)
+            pxrun_cell = _asyncify(pxrun_cell)
         self.shell.run_cell = pxrun_cell
 
         pxrun_nodes = self.pxrun_nodes
         if _iscoroutinefunction(self.shell.run_ast_nodes):
             # original is a coroutine,
             # wrap ours in a coroutine
-            pxrun_nodes = coroutine(pxrun_nodes)
+            pxrun_nodes = _asyncify(pxrun_nodes)
         self.shell.run_ast_nodes = pxrun_nodes
 
         self._autopx = True

ipyparallel/controller/hub.py

@@ -7,6 +7,7 @@
 # Distributed under the terms of the Modified BSD License.
 from __future__ import print_function
 
+import inspect
 import json
 import os
 import sys
@@ -20,8 +21,6 @@
 from jupyter_client.jsonutil import parse_date
 from jupyter_client.session import Session
 from tornado import ioloop
-from tornado.gen import coroutine
-from tornado.gen import maybe_future
 from traitlets import Any
 from traitlets import Bytes
 from traitlets import Dict
@@ -319,8 +318,7 @@ def dispatch_monitor_traffic(self, msg):
             self.log.error("Unrecognized monitor topic: %r", switch)
 
     @util.log_errors
-    @coroutine
-    def dispatch_query(self, msg):
+    async def dispatch_query(self, msg):
         """Route registration requests and queries from clients."""
         try:
             idents, msg = self.session.feed_identities(msg)
@@ -357,8 +355,8 @@ def dispatch_query(self, msg):
 
         try:
             f = handler(idents, msg)
-            if f:
-                yield maybe_future(f)
+            if f and inspect.isawaitable(f):
+                await f
         except Exception:
             content = error.wrap_exception()
             self.log.error("Error handling request: %r", msg_type, exc_info=True)
@@ -1450,16 +1448,15 @@ def db_query(self, client_id, msg):
             buffers=buffers,
         )
 
-    @coroutine
-    def become_dask(self, client_id, msg):
+    async def become_dask(self, client_id, msg):
         """Start a dask.distributed Scheduler."""
         if self.distributed_scheduler is None:
             kwargs = msg['content'].get('scheduler_args', {})
             self.log.info("Becoming dask.distributed scheduler: %s", kwargs)
             from distributed import Scheduler
 
             self.distributed_scheduler = scheduler = Scheduler(**kwargs)
-            yield scheduler.start()
+            await scheduler.start()
         content = {
             'status': 'ok',
             'ip': self.distributed_scheduler.ip,