如何捕获gen.Task内部的异常？

Question

我使用的是python 2.7，龙卷风4.5以下代码不起作用： except 块不会被触发。我不明白为什么？@gen.coroutinedef co_do_thing():  yield gen.Task(do_thing)def do_thing(callback):  try:    a, b = ...    result = maybe_throw(a, b, callback)  except Exception as e:    # this block is not called    if a:      raise ApiError("called with A")    elif b:      raise ApiError("called with B")    else:      raise edef maybe_throw(arg1, arg2, callback):  if random.random() < 0.5:    raise AssertionError("yikes")  callback("done")co_do_thing相反，我可以在调用周围捕获异常gen.Task；但后来我不知道我如何调用的上下文maybe_throw。maybe_throw就我而言，引发较低级别的异常并让调用者根据输入将其转换为人类可读的错误更有意义。我是否只需要重构它以在较低级别调用 gen.Task ？那会很烦人：/

Answer 1

当我测试它似乎有效时，会引发异常。下面是简单的测试套件：

import q  # q.py is the file with question's code

import unittest

from mock import patch, Mock

from tornado.testing import gen_test, AsyncTestCase

class MyTest(AsyncTestCase):

    def setUp(self):

        self.mock_random = patch('q.random').start()

        AsyncTestCase.setUp(self)

    def tearDown(self):

        AsyncTestCase.tearDown(self)

        patch.stopall()

    @gen_test

    def test_no_error(self):

        self.mock_random.return_value = 0.7

        res = yield q.co_do_thing()

        self.assertEqual(res, 'done')

    @gen_test

    def test_exception(self):

        self.mock_random.return_value = 0.1

        with self.assertRaises(Exception) as ctx:

            yield q.co_do_thing()

        self.assertEqual(ctx.exception.message, 'called with A')

if __name__ == '__main__':

    unittest.main()

测试通过了：

..

----------------------------------------------------------------------

Ran 2 tests in 0.002s

OK

这是q.py，我添加了 return 语句来测试它。

from random import random

from tornado import gen

@gen.coroutine

def co_do_thing():

    res = yield gen.Task(do_thing)

    # added: to enable to test it meaningfully

    raise gen.Return(res)

def do_thing(callback):

    try:

        a, b = 22, 33

        result = maybe_throw(a, b, callback)

    except Exception as e:

        if a:

            raise Exception("called with A")

        elif b:

            raise Exception("called with B")

        else:

            raise e

def maybe_throw(arg1, arg2, callback):

    if random() < 0.5:

        raise AssertionError("yikes")

    callback("done")