在其消费者中处理生成器异常答案

【问题标题】：Handle generator exceptions in its consumer在其消费者中处理生成器异常
【发布时间】：2012-07-07 04:32:37
【问题描述】：

这是Handle an exception thrown in a generator 的后续内容，讨论了一个更普遍的问题。

我有一个读取不同格式数据的函数。所有格式都是面向行或面向记录的，并且对于每种格式都有一个专用的解析功能，作为生成器实现。所以主读取函数得到一个输入和一个生成器，它从输入中读取其各自的格式并将记录传递回主函数：

def read(stream, parsefunc):
    for record in parsefunc(stream):
        do_stuff(record)

parsefunc 类似于：

def parsefunc(stream):
    while not eof(stream):
        rec = read_record(stream)
        do some stuff
        yield rec

我面临的问题是，虽然parsefunc 可以抛出异常（例如，从流中读取时），但它不知道如何处理它。负责处理异常的函数是主要的read 函数。请注意，异常发生在每条记录的基础上，因此即使一条记录失败，生成器也应继续其工作并返回记录，直到整个流耗尽。

在上一个问题中，我尝试将next(parsefunc) 放在try 块中，但事实证明，这是行不通的。所以我必须将try-except 添加到parsefunc 本身，然后以某种方式将异常传递给消费者：

def parsefunc(stream):
    while not eof(stream):
        try:
            rec = read_record()
            yield rec
        except Exception as e:
            ?????

我不太愿意这样做，因为

在不打算处理任何异常的函数中使用 try 是没有意义的
我不清楚如何将异常传递给消费函数
会有很多格式和很多 parsefunc's，我不想用太多的帮助代码弄乱它们。

有没有人建议更好的架构？

Google 员工注意事项：除了最佳答案之外，请注意 senderle's 和 Jon's 的帖子 - 非常聪明和有见地的东西。

【问题讨论】：

讨论一个设计通常需要相当详细的系统知识，所以在这里似乎很难给出建议。我的第一个想法是取消parsefunc()，并将read_record 传递给read()。这样，可以在调用read_record() 之后立即处理异常。
我同意@SvenMarnach - 我阅读它的方式可能过于复杂 - 即，parsefunc() 负责消费记录，但是 - 如何适当parsefunc() 选择用于流的其余部分？
@SvenMarnach：实际上，我在parsefunc 中描述为do some stuff 的代码相当大，所以我不能简单地摆脱它。
@JonClements：调用read 的人都知道数据的格式，并决定使用哪个解析器。
@eof 定义在哪里？

标签： python exception-handling generator

【解决方案1】：

你可以在 parsefunc 中返回一个记录和异常的元组，让消费者函数决定如何处理异常：

import random

def get_record(line):
  num = random.randint(0, 3)
  if num == 3:
    raise Exception("3 means danger")
  return line


def parsefunc(stream):
  for line in stream:
    try:
      rec = get_record(line)
    except Exception as e:
      yield (None, e)
    else:
      yield (rec, None)

if __name__ == '__main__':
  with open('temp.txt') as f:
    for rec, e in parsefunc(f):
      if e:
        print "Got an exception %s" % e
      else:
        print "Got a record %s" % rec

【讨论】：

谢谢！这与我最终做的类似。

【解决方案2】：

更深入地思考在更复杂的情况下会发生什么表明 Python 选择避免从生成器中冒泡异常。

如果我从流对象中得到一个 I/O 错误，那么在不以某种方式重置生成器本地结构的情况下，能够简单地恢复并继续读取的几率会很低。我必须以某种方式使自己与阅读过程相协调才能继续：跳过垃圾，推回部分数据，重置一些不完整的内部跟踪结构等。

只有生成器有足够的上下文才能正确执行此操作。即使您可以保留生成器上下文，让外部块处理异常也会完全违反得墨忒耳法则。周围块需要重置和继续前进的所有重要信息都在生成器函数的局部变量中！获取或传递这些信息，尽管可能，但令人作呕。

产生的异常几乎总是在清理之后抛出，在这种情况下，读取器-生成器将已经有一个内部异常块。非常努力地在脑死亡简单的情况下保持这种清洁，只是让它在几乎所有现实情况下都崩溃是愚蠢的。所以只要在生成器中拥有try，无论如何，在任何复杂情况下，您都将需要except 块的主体。

如果异常情况看起来像异常，而不像返回值，那就太好了。所以我会添加一个中间适配器来实现这一点：生成器将产生数据或异常，如果适用，适配器将重新引发异常。适配器应该在 for 循环中首先被调用，这样我们就可以选择在循环中捕获它并清理以继续，或者跳出循环以捕获它并放弃该过程。而且我们应该在设置周围放置某种蹩脚的包装器，以表明技巧正在发生，并在函数正在适应时强制调用适配器。

这样一来，每个层都会显示它具有要处理的上下文的错误，但代价是适配器有点侵入性（而且可能也很容易忘记）。

所以我们会：

def read(stream, parsefunc):
  try:
    for source in frozen(parsefunc(stream)):
      try:
        record = source.thaw()
        do_stuff(record)
      except Exception, e:
        log_error(e)
        if not is_recoverable(e):
          raise
        recover()
  except Exception, e:
    properly_give_up()
  wrap_up()

（其中两个 try 块是可选的。）

适配器看起来像：

class Frozen(object):
  def __init__(self, item):
    self.value = item
  def thaw(self):
    if isinstance(value, Exception):
      raise value
    return value

def frozen(generator):
    for item in generator:
       yield Frozen(item)

parsefunc 看起来像：

def parsefunc(stream):
  while not eof(stream):
    try:
       rec = read_record(stream)
       do_some_stuff()
       yield rec
    except Exception, e:
       properly_skip_record_or_prepare_retry()
       yield e

为了更难忘记适配器，我们还可以将 freeze 从一个函数更改为 parsefunc 上的装饰器。

def frozen_results(func):
  def freezer(__func = func, *args, **kw):
    for item in __func(*args, **kw):
       yield Frozen(item)
  return freezer

在这种情况下，我们将声明：

@frozen_results
def parsefunc(stream):
  ...

我们显然不会费心声明frozen，或者将它包装在对parsefunc的调用中。

【讨论】：

【解决方案3】：

如果不了解该系统的更多信息，我认为很难判断哪种方法最有效。但是，没有人建议的一种选择是使用回调。鉴于只有 read 知道如何处理异常，这样的方法可能有用吗？

def read(stream, parsefunc):
    some_closure_data = {}

    def error_callback_1(e):
        manipulate(some_closure_data, e)
    def error_callback_2(e):
        transform(some_closure_data, e)

    for record in parsefunc(stream, error_callback_1):
        do_stuff(record)

然后，在parsefunc：

def parsefunc(stream, error_callback):
    while not eof(stream):
        try:
            rec = read_record()
            yield rec
        except Exception as e:
            error_callback(e)

我在这里对可变局部使用了闭包；你也可以定义一个类。另请注意，您可以通过回调内部的sys.exc_info() 访问traceback 信息。

另一个有趣的方法可能是使用send。这会有点不同。基本上，read 可以检查yield 的结果，而不是定义回调，做很多复杂的逻辑，send 一个替代值，然后生成器将重新生成（或做其他事情） .这有点异国情调，但我想我会提到它以防它有用：

>>> def parsefunc(it):
...     default = None
...     for x in it:
...         try:
...             rec = float(x)
...         except ValueError as e:
...             default = yield e
...             yield default
...         else:
...             yield rec
... 
>>> parsed_values = parsefunc(['4', '6', '5', '5h', '22', '7'])
>>> for x in parsed_values:
...     if isinstance(x, ValueError):
...         x = parsed_values.send(0.0)
...     print x
... 
4.0
6.0
5.0
0.0
22.0
7.0

就其本身而言，这有点无用（“为什么不直接从read 打印默认值？”你可能会问），但你可以在生成器中使用default 做更复杂的事情，重置值，继续后退一步，以此类推。您甚至可以根据收到的错误等待在 this 点发送回调。但请注意，sys.exc_info() 会在生成器yields 后立即清除，因此如果您需要访问回溯，则必须从sys.exc_info() 发送所有内容。

以下是您如何组合这两个选项的示例：

import string
digits = set(string.digits)

def digits_only(v):
    return ''.join(c for c in v if c in digits)

def parsefunc(it):
    default = None
    for x in it:
        try:
            rec = float(x)
        except ValueError as e:
            callback = yield e
            yield float(callback(x))
        else:
            yield rec

parsed_values = parsefunc(['4', '6', '5', '5h', '22', '7'])
for x in parsed_values:
    if isinstance(x, ValueError):
        x = parsed_values.send(digits_only)
    print x

【讨论】：

确实很有趣，向我展示了 python 编程的全新一面。非常感谢！

【解决方案4】：

一个可能的设计示例：

from StringIO import StringIO
import csv

blah = StringIO('this,is,1\nthis,is\n')

def parse_csv(stream):
    for row in csv.reader(stream):
        try:
            yield int(row[2])
        except (IndexError, ValueError) as e:
            pass # don't yield but might need something
        # All others have to go up a level - so it wasn't parsable
        # So if it's an IOError you know why, but this needs to catch
        # exceptions potentially, just let the major ones propogate

for record in parse_csv(blah):
    print record

【讨论】：

【解决方案5】：

我喜欢Frozen 的给定答案。基于这个想法，我想出了这个，解决了我还不喜欢的两个方面。首先是写下来所需的模式。第二个是在产生异常时丢失堆栈跟踪。我尽我所能通过使用尽可能好的装饰器来解决第一个问题。我尝试使用 sys.exc_info() 而不是单独使用异常来保留堆栈跟踪。

我的生成器通常（即没有应用我的东西）看起来像这样：

def generator():
  def f(i):
    return float(i) / (3 - i)
  for i in range(5):
    yield f(i)

如果我可以将其转换为使用内部函数来确定要产生的值，我可以应用我的方法：

def generator():
  def f(i):
    return float(i) / (3 - i)
  for i in range(5):
    def generate():
      return f(i)
    yield generate()

这还没有改变任何东西，像这样调用它会引发错误并带有适当的堆栈跟踪：

for e in generator():
  print e

现在，应用我的装饰器，代码将如下所示：

@excepterGenerator
def generator():
  def f(i):
    return float(i) / (3 - i)
  for i in range(5):
    @excepterBlock
    def generate():
      return f(i)
    yield generate()

光学上没有太大变化。而且你仍然可以像之前使用的版本一样使用它：

for e in generator():
  print e

并且在调用时您仍然可以得到正确的堆栈跟踪。（现在只剩下一帧了。）

但现在你也可以这样使用它：

it = generator()
while it:
  try:
    for e in it:
      print e
  except Exception as problem:
    print 'exc', problem

通过这种方式，您可以在消费者中处理生成器中引发的任何异常，而无需过多的语法麻烦，也不会丢失堆栈跟踪。

装饰器的拼写如下：

import sys

def excepterBlock(code):
  def wrapper(*args, **kwargs):
    try:
      return (code(*args, **kwargs), None)
    except Exception:
      return (None, sys.exc_info())
  return wrapper

class Excepter(object):
  def __init__(self, generator):
    self.generator = generator
    self.running = True
  def next(self):
    try:
      v, e = self.generator.next()
    except StopIteration:
      self.running = False
      raise
    if e:
      raise e[0], e[1], e[2]
    else:
      return v
  def __iter__(self):
    return self
  def __nonzero__(self):
    return self.running

def excepterGenerator(generator):
  return lambda *args, **kwargs: Excepter(generator(*args, **kwargs))

【讨论】：

【解决方案6】：

（我回答了 OP 中链接的另一个问题，但我的回答也适用于这种情况）

我需要解决这个问题几次，并在搜索其他人做了什么后发现了这个问题。

一个选项（可能需要稍微重构一下）是简单地创建一个错误处理生成器，并在生成器中创建throw 异常（到另一个错误处理生成器）而不是raise 它。

下面是错误处理生成器函数的样子：

def err_handler():
    # a generator for processing errors
    while True:
        try:
            # errors are thrown to this point in function
            yield
        except Exception1:
            handle_exc1()
        except Exception2:
            handle_exc2()
        except Exception3:
            handle_exc3()
        except Exception:
            raise

为parsefunc 函数提供了一个额外的handler 参数，因此它可以放置错误：

def parsefunc(stream, handler):
    # the handler argument fixes errors/problems separately
    while not eof(stream):
        try:
            rec = read_record(stream)
            do some stuff
            yield rec
        except Exception as e:
            handler.throw(e)
    handler.close()

现在几乎只使用原始的read 函数，但现在使用错误处理程序：

def read(stream, parsefunc):
    handler = err_handler()
    for record in parsefunc(stream, handler):
        do_stuff(record)

这并不总是最好的解决方案，但它肯定是一种选择，而且相对容易理解。

【讨论】：

【解决方案7】：

关于将异常从生成器传播到消费函数的观点，您可以尝试使用错误代码（错误代码集）来指示错误。虽然不优雅，但这是您能想到的一种方法。

例如，在下面的代码中，会在您期望的地方产生类似 -1 的值一组正整数将向调用函数发出信号，表明存在一个错误。

In [1]: def f():
  ...:     yield 1
  ...:     try:
  ...:         2/0
  ...:     except ZeroDivisionError,e:
  ...:         yield -1
  ...:     yield 3
  ...:     


In [2]: g = f()

In [3]: next(g)
Out[3]: 1

In [4]: next(g)
Out[4]: -1

In [5]: next(g)
Out[5]: 3

【讨论】：

【解决方案8】：

实际上，生成器在几个方面都非常有限。您发现了一个：引发异常不是其 API 的一部分。

你可以看看 Stackless Python 的东西，比如 greenlets 或 coroutines，它们提供了更多的灵活性；但是在这里深入研究有点超出了范围。

【讨论】：