Python中的异步异常处理

Question

我有以下代码使用asyncio和aiohttp发出异步HTTP请求.

import sys
import asyncio
import aiohttp

@asyncio.coroutine
def get(url):
    try:
        print('GET %s' % url)
        resp = yield from aiohttp.request('GET', url)
    except Exception as e:
        raise Exception("%s has error '%s'" % (url, e))
    else:
        if resp.status >= 400:
            raise Exception("%s has error '%s: %s'" % (url, resp.status, resp.reason))

    return (yield from resp.text())

@asyncio.coroutine
def fill_data(run):
    url = 'http://www.google.com/%s' % run['name']
    run['data'] = yield from get(url)

def get_runs():
    runs = [ {'name': 'one'}, {'name': 'two'} ]
    loop = asyncio.get_event_loop()
    task = asyncio.wait([fill_data(r) for r in runs])
    loop.run_until_complete(task)   
    return runs

try:
    get_runs()
except Exception as e:
    print(repr(e))
    sys.exit(1)

出于某种原因,get函数内部引发的异常未被捕获:

Future/Task exception was never retrieved
Traceback (most recent call last):
  File "site-packages/asyncio/tasks.py", line 236, in _step
    result = coro.send(value)
  File "mwe.py", line 25, in fill_data
    run['data'] = yield from get(url)
  File "mwe.py", line 17, in get
    raise Exception("%s has error '%s: %s'" % (url, resp.status, resp.reason))
Exception: http://www.google.com/two has error '404: Not Found'

那么,处理由couroutines引发的异常的正确方法是什么？

Answer 1

asyncio.wait实际上并没有消耗Futures传递给它,只是等待它们完成,然后返回Future对象:

协程 asyncio.wait(futures, *, loop=None, timeout=None, return_when=ALL_COMPLETED)

等待序列期货给出的期货和协程对象完成.协同程序将包含在任务中.返回两组Future:(完成,待定).

直到你实际上列表中yield from的项目done,他们将保持未消耗.由于您的程序在不消耗期货的情况下退出,因此您会看到"异常从未检索过"消息.

对于您的用例,使用它可能更有意义asyncio.gather,它实际上会消耗每个Future,然后返回一个Future聚合所有结果的单个(或者引发Exception输入列表中未来抛出的第一个).

def get_runs():
    runs = [ {'name': 'one'}, {'name': 'two'} ]
    loop = asyncio.get_event_loop()
    tasks = asyncio.gather(*[fill_data(r) for r in runs])
    loop.run_until_complete(tasks)
    return runs

输出:

GET http://www.google.com/two
GET http://www.google.com/one
Exception("http://www.google.com/one has error '404: Not Found'",)

请注意,asyncio.gather实际上,当其中一个期货引发异常时,您可以自定义其行为; 默认行为是引发它命中的第一个异常,但它也可以只返回输出列表中的每个异常对象:

asyncio.gather(*coros_or_futures, loop=None, return_exceptions=False)

返回给定协程对象或期货的未来聚合结果.

所有期货必须共享相同的事件循环.如果所有任务都成功完成,则返回的将来的结果是结果列表(按原始序列的顺序,不一定是结果到达的顺序).如果return_exceptions是True,则将任务中的异常视为与成功结果相同,并在结果列表中收集; 否则,第一个引发的异常将立即传播到返回的未来.

Answer 2

要在回调中调试或“处理”异常：

返回一些结果或引发异常的协程：

@asyncio.coroutine
def async_something_entry_point(self):
    try:
        return self.real_stuff_which_throw_exceptions()
    except:
        raise Exception(some_identifier_here + ' ' + traceback.format_exc())

和回调：

def callback(self, future: asyncio.Future):
    exc = future.exception()
    if exc:
        # Handle wonderful empty TimeoutError exception
        if type(exc) == TimeoutError:
            self.logger('<Some id here> callback exception TimeoutError')
        else:
            self.logger("<Some id here> callback exception " + str(exc))

    # store your result where you want
    self.result.append(
        future.result()
    )