]> git.ipfire.org Git - thirdparty/tornado.git/commitdiff
Improve introspection of coroutines 1890/head
authorAntoine Pitrou <antoine@python.org>
Tue, 15 Nov 2016 11:35:29 +0000 (12:35 +0100)
committerAntoine Pitrou <antoine@python.org>
Tue, 15 Nov 2016 11:35:29 +0000 (12:35 +0100)
docs/gen.rst
tornado/gen.py
tornado/test/gen_test.py

index 8e867ed0e7b57b3fb257cd45a850f1eec5c7e0c8..52c7f55d9bb8086eeb617d2a8f9fc50e9914b6cf 100644 (file)
@@ -50,6 +50,8 @@
 
    .. autofunction:: maybe_future
 
+   .. autofunction:: is_coroutine_function
+
    Legacy interface
    ----------------
 
index 73f9ba10a36f133a7fd940bb7f7022ca49e97738..d7df3b52f3cb4686d8e8da951a783f7e223a3963 100644 (file)
@@ -273,10 +273,11 @@ def _make_coroutine_wrapper(func, replace_callback):
     """
     # On Python 3.5, set the coroutine flag on our generator, to allow it
     # to be used with 'await'.
+    wrapped = func
     if hasattr(types, 'coroutine'):
         func = types.coroutine(func)
 
-    @functools.wraps(func)
+    @functools.wraps(wrapped)
     def wrapper(*args, **kwargs):
         future = TracebackFuture()
 
@@ -328,9 +329,19 @@ def _make_coroutine_wrapper(func, replace_callback):
                     future = None
         future.set_result(result)
         return future
+
+    wrapper.__wrapped__ = wrapped
+    wrapper.__tornado_coroutine__ = True
     return wrapper
 
 
+def is_coroutine_function(func):
+    """Return whether *func* is a coroutine function, i.e. a function
+    wrapped with `~.gen.coroutine`.
+    """
+    return getattr(func, '__tornado_coroutine__', False)
+
+
 class Return(Exception):
     """Special exception to return a value from a `coroutine`.
 
index bfaba5678c554ecd5a9c18d09f8126bcf8264cc3..8bbfc5fa8f2ff16cc633306557bae35fb7252e51 100644 (file)
@@ -657,6 +657,28 @@ class GenCoroutineTest(AsyncTestCase):
         super(GenCoroutineTest, self).tearDown()
         assert self.finished
 
+    def test_attributes(self):
+        self.finished = True
+
+        def f():
+            yield gen.moment
+
+        coro = gen.coroutine(f)
+        self.assertEqual(coro.__name__, f.__name__)
+        self.assertEqual(coro.__module__, f.__module__)
+        self.assertIs(coro.__wrapped__, f)
+
+    def test_is_coroutine_function(self):
+        self.finished = True
+
+        def f():
+            yield gen.moment
+
+        coro = gen.coroutine(f)
+        self.assertFalse(gen.is_coroutine_function(f))
+        self.assertTrue(gen.is_coroutine_function(coro))
+        self.assertFalse(gen.is_coroutine_function(coro()))
+
     @gen_test
     def test_sync_gen_return(self):
         @gen.coroutine