How to catch exceptions inside gen.Task?

Question

I'm on python 2.7, tornado 4.5

The following code doesn't work: the except block doesn't get triggered. I don't understand why?

@gen.coroutine
def 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 e

def maybe_throw(arg1, arg2, callback):
  if random.random() < 0.5:
    raise AssertionError("yikes")
  callback("done")

Instead, I can catch the exception in co_do_thing around the call to gen.Task; but then I don't have the context of how I called maybe_throw. In my case, it makes more sense for maybe_throw to raise a lower-level exception, and for the caller to convert that to a human-readable error depending on the inputs.

Do I just need to refactor this to call gen.Task at a lower level? That would be annoying :/

Answer 1

As I tested it seems to work, a exception is raised. Below simple test suite:

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()

And tests passed:

..
----------------------------------------------------------------------
Ran 2 tests in 0.002s

OK

And here is q.py, I've added return statement to test it.

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")