我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用tornado.gen.engine()。
def test_yield_in_with(self): @gen.engine def f(): self.callback = yield gen.Callback('a') with StackContext(functools.partial(self.context, 'c1')): # This yield is a problem: the generator will be suspended # and the StackContext's __exit__ is not called yet, so # the context will be left on _state.contexts for anything # that runs before the yield resolves. yield gen.Wait('a') with self.assertRaises(StackContextInconsistentError): f() self.wait() # Cleanup: to avoid GC warnings (which for some reason only seem # to show up on py33-asyncio), invoke the callback (which will do # nothing since the gen.Runner is already finished) and delete it. self.callback() del self.callback
def test_stack_context_leak_exception(self): # same as previous, but with a function that exits with an exception @gen.engine def inner(callback): yield gen.Task(self.io_loop.add_callback) 1 / 0 @gen.engine def outer(): for i in range(10): try: yield gen.Task(inner) except ZeroDivisionError: pass stack_increase = len(stack_context._state.contexts) - initial_stack_depth self.assertTrue(stack_increase <= 2) self.stop() initial_stack_depth = len(stack_context._state.contexts) self.run_gen(outer)
def test_task_refcounting(self): # On CPython, tasks and their arguments should be released immediately # without waiting for garbage collection. @gen.engine def f(): class Foo(object): pass arg = Foo() self.arg_ref = weakref.ref(arg) task = gen.Task(self.io_loop.add_callback, arg=arg) self.task_ref = weakref.ref(task) yield task self.stop() self.run_gen(f) self.assertIs(self.arg_ref(), None) self.assertIs(self.task_ref(), None)
def test_future_traceback(self): @return_future @gen.engine def f(callback): yield gen.Task(self.io_loop.add_callback) try: 1 / 0 except ZeroDivisionError: self.expected_frame = traceback.extract_tb( sys.exc_info()[2], limit=1)[0] raise try: yield f() self.fail("didn't get expected exception") except ZeroDivisionError: tb = traceback.extract_tb(sys.exc_info()[2]) self.assertIn(self.expected_frame, tb) # The following series of classes demonstrate and test various styles # of use, with and without generators and futures.
def test_yield_in_with_exception_stack_context(self): # As above, but with ExceptionStackContext instead of StackContext. @gen.engine def f(): with ExceptionStackContext(lambda t, v, tb: False): yield gen.Task(self.io_loop.add_callback) with self.assertRaises(StackContextInconsistentError): f() self.wait()
def test_no_yield(self): @gen.engine def f(): self.stop() self.run_gen(f)
def test_inline_cb(self): @gen.engine def f(): (yield gen.Callback("k1"))() res = yield gen.Wait("k1") self.assertTrue(res is None) self.stop() self.run_gen(f)
def test_ioloop_cb(self): @gen.engine def f(): self.io_loop.add_callback((yield gen.Callback("k1"))) yield gen.Wait("k1") self.stop() self.run_gen(f)
def test_exception_phase2(self): @gen.engine def f(): self.io_loop.add_callback((yield gen.Callback("k1"))) yield gen.Wait("k1") 1 / 0 self.assertRaises(ZeroDivisionError, self.run_gen, f)
def test_exception_in_task_phase1(self): def fail_task(callback): 1 / 0 @gen.engine def f(): try: yield gen.Task(fail_task) raise Exception("did not get expected exception") except ZeroDivisionError: self.stop() self.run_gen(f)
def test_exception_in_task_phase2(self): # This is the case that requires the use of stack_context in gen.engine def fail_task(callback): self.io_loop.add_callback(lambda: 1 / 0) @gen.engine def f(): try: yield gen.Task(fail_task) raise Exception("did not get expected exception") except ZeroDivisionError: self.stop() self.run_gen(f)
def test_with_arg(self): @gen.engine def f(): (yield gen.Callback("k1"))(42) res = yield gen.Wait("k1") self.assertEqual(42, res) self.stop() self.run_gen(f)
def test_with_arg_tuple(self): @gen.engine def f(): (yield gen.Callback((1, 2)))((3, 4)) res = yield gen.Wait((1, 2)) self.assertEqual((3, 4), res) self.stop() self.run_gen(f)
def test_key_reuse_tuple(self): @gen.engine def f(): yield gen.Callback((1, 2)) yield gen.Callback((1, 2)) self.stop() self.assertRaises(gen.KeyReuseError, self.run_gen, f)
def test_key_mismatch(self): @gen.engine def f(): yield gen.Callback("k1") yield gen.Wait("k2") self.stop() self.assertRaises(gen.UnknownKeyError, self.run_gen, f)
def test_key_mismatch_tuple(self): @gen.engine def f(): yield gen.Callback((1, 2)) yield gen.Wait((2, 3)) self.stop() self.assertRaises(gen.UnknownKeyError, self.run_gen, f)
def test_leaked_callback(self): @gen.engine def f(): yield gen.Callback("k1") self.stop() self.assertRaises(gen.LeakedCallbackError, self.run_gen, f)
def test_leaked_callback_tuple(self): @gen.engine def f(): yield gen.Callback((1, 2)) self.stop() self.assertRaises(gen.LeakedCallbackError, self.run_gen, f)
def test_bogus_yield(self): @gen.engine def f(): yield 42 self.assertRaises(gen.BadYieldError, self.run_gen, f)
def test_bogus_yield_tuple(self): @gen.engine def f(): yield (1, 2) self.assertRaises(gen.BadYieldError, self.run_gen, f)
def test_reuse(self): @gen.engine def f(): self.io_loop.add_callback((yield gen.Callback(0))) yield gen.Wait(0) self.stop() self.run_gen(f) self.run_gen(f)
def test_task(self): @gen.engine def f(): yield gen.Task(self.io_loop.add_callback) self.stop() self.run_gen(f)
def test_wait_all(self): @gen.engine def f(): (yield gen.Callback("k1"))("v1") (yield gen.Callback("k2"))("v2") results = yield gen.WaitAll(["k1", "k2"]) self.assertEqual(results, ["v1", "v2"]) self.stop() self.run_gen(f)
def test_resume_after_exception_in_yield(self): @gen.engine def f(): try: yield gen.Wait("k1") raise Exception("did not get expected exception") except gen.UnknownKeyError: pass (yield gen.Callback("k2"))("v2") self.assertEqual((yield gen.Wait("k2")), "v2") self.stop() self.run_gen(f)
def test_orphaned_callback(self): @gen.engine def f(): self.orphaned_callback = yield gen.Callback(1) try: self.run_gen(f) raise Exception("did not get expected exception") except gen.LeakedCallbackError: pass self.orphaned_callback()
def test_multi(self): @gen.engine def f(): (yield gen.Callback("k1"))("v1") (yield gen.Callback("k2"))("v2") results = yield [gen.Wait("k1"), gen.Wait("k2")] self.assertEqual(results, ["v1", "v2"]) self.stop() self.run_gen(f)
def test_multi_dict(self): @gen.engine def f(): (yield gen.Callback("k1"))("v1") (yield gen.Callback("k2"))("v2") results = yield dict(foo=gen.Wait("k1"), bar=gen.Wait("k2")) self.assertEqual(results, dict(foo="v1", bar="v2")) self.stop() self.run_gen(f) # The following tests explicitly run with both gen.Multi # and gen.multi_future (Task returns a Future, so it can be used # with either).
def test_multi_yieldpoint_delayed(self): @gen.engine def f(): # callbacks run at different times responses = yield gen.Multi([ gen.Task(self.delay_callback, 3, arg="v1"), gen.Task(self.delay_callback, 1, arg="v2"), ]) self.assertEqual(responses, ["v1", "v2"]) self.stop() self.run_gen(f)
def test_multi_future_delayed(self): @gen.engine def f(): # callbacks run at different times responses = yield gen.multi_future([ gen.Task(self.delay_callback, 3, arg="v1"), gen.Task(self.delay_callback, 1, arg="v2"), ]) self.assertEqual(responses, ["v1", "v2"]) self.stop() self.run_gen(f)
def test_multi_future_dict_delayed(self): @gen.engine def f(): # callbacks run at different times responses = yield gen.multi_future(dict( foo=gen.Task(self.delay_callback, 3, arg="v1"), bar=gen.Task(self.delay_callback, 1, arg="v2"), )) self.assertEqual(responses, dict(foo="v1", bar="v2")) self.stop() self.run_gen(f)
def test_arguments(self): @gen.engine def f(): (yield gen.Callback("noargs"))() self.assertEqual((yield gen.Wait("noargs")), None) (yield gen.Callback("1arg"))(42) self.assertEqual((yield gen.Wait("1arg")), 42) (yield gen.Callback("kwargs"))(value=42) result = yield gen.Wait("kwargs") self.assertTrue(isinstance(result, gen.Arguments)) self.assertEqual(((), dict(value=42)), result) self.assertEqual(dict(value=42), result.kwargs) (yield gen.Callback("2args"))(42, 43) result = yield gen.Wait("2args") self.assertTrue(isinstance(result, gen.Arguments)) self.assertEqual(((42, 43), {}), result) self.assertEqual((42, 43), result.args) def task_func(callback): callback(None, error="foo") result = yield gen.Task(task_func) self.assertTrue(isinstance(result, gen.Arguments)) self.assertEqual(((None,), dict(error="foo")), result) self.stop() self.run_gen(f)
def test_stack_context_leak(self): # regression test: repeated invocations of a gen-based # function should not result in accumulated stack_contexts def _stack_depth(): head = stack_context._state.contexts[1] length = 0 while head is not None: length += 1 head = head.old_contexts[1] return length @gen.engine def inner(callback): yield gen.Task(self.io_loop.add_callback) callback() @gen.engine def outer(): for i in range(10): yield gen.Task(inner) stack_increase = _stack_depth() - initial_stack_depth self.assertTrue(stack_increase <= 2) self.stop() initial_stack_depth = _stack_depth() self.run_gen(outer)
def test_sync_raise_return(self): # gen.Return is allowed in @gen.engine, but it may not be used # to return a value. @gen.engine def f(): self.stop(42) raise gen.Return() result = self.run_gen(f) self.assertEqual(result, 42)
def test_async_raise_return(self): @gen.engine def f(): yield gen.Task(self.io_loop.add_callback) self.stop(42) raise gen.Return() result = self.run_gen(f) self.assertEqual(result, 42)
def test_sync_raise_return_value(self): @gen.engine def f(): raise gen.Return(42) with self.assertRaises(gen.ReturnValueIgnoredError): self.run_gen(f)
def test_sync_raise_return_value_tuple(self): @gen.engine def f(): raise gen.Return((1, 2)) with self.assertRaises(gen.ReturnValueIgnoredError): self.run_gen(f)
def test_async_raise_return_value(self): @gen.engine def f(): yield gen.Task(self.io_loop.add_callback) raise gen.Return(42) with self.assertRaises(gen.ReturnValueIgnoredError): self.run_gen(f)
def test_return_value(self): # It is an error to apply @gen.engine to a function that returns # a value. @gen.engine def f(): return 42 with self.assertRaises(gen.ReturnValueIgnoredError): self.run_gen(f)
def test_return_value_tuple(self): # It is an error to apply @gen.engine to a function that returns # a value. @gen.engine def f(): return (1, 2) with self.assertRaises(gen.ReturnValueIgnoredError): self.run_gen(f)
def get(self): if self.get_argument("oauth_token", None): user = yield self.get_authenticated_user() self.finish(user) else: # Old style: with @gen.engine we can ignore the Future from # authorize_redirect. self.authorize_redirect()
