From e068d99cbf0b8a6661602a4b2320fb18b4a0e687 Mon Sep 17 00:00:00 2001 From: Antoine Pitrou Date: Tue, 15 Nov 2016 12:35:29 +0100 Subject: [PATCH] Improve introspection of coroutines --- docs/gen.rst | 2 ++ tornado/gen.py | 13 ++++++++++++- tornado/test/gen_test.py | 22 ++++++++++++++++++++++ 3 files changed, 36 insertions(+), 1 deletion(-) diff --git a/docs/gen.rst b/docs/gen.rst index 8e867ed0..52c7f55d 100644 --- a/docs/gen.rst +++ b/docs/gen.rst @@ -50,6 +50,8 @@ .. autofunction:: maybe_future + .. autofunction:: is_coroutine_function + Legacy interface ---------------- diff --git a/tornado/gen.py b/tornado/gen.py index 73f9ba10..d7df3b52 100644 --- a/tornado/gen.py +++ b/tornado/gen.py @@ -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`. diff --git a/tornado/test/gen_test.py b/tornado/test/gen_test.py index bfaba567..8bbfc5fa 100644 --- a/tornado/test/gen_test.py +++ b/tornado/test/gen_test.py @@ -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