python - 在其消费者中处理生成器异常

Question

这是处理生成器中抛出的异常的后续内容，并讨论了一个更普遍的问题。

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

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可以抛出异常（例如从流中读取时），但它不知道如何处理它。负责处理异常的函数是mainread函数。请注意，异常发生在每条记录的基础上，因此即使一条记录失败，生成器也应继续其工作并返回记录，直到整个流耗尽。

在上一个问题中，我尝试放入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，我不想用太多的帮助代码弄乱它们。

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

googlers 的注意事项：除了最佳答案之外，还要注意senderle和Jon 的帖子——非常聪明和有见地的东西。

Answer 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

Answer 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.

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

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

另一种有趣的方法可能是使用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在生成器内部做更复杂的事情，重置值，后退一步，等等。您甚至可以根据收到的错误等待此时发送回调。但请注意，sys.exc_info()一旦生成器 s，它就会被清除，因此如果您需要访问回溯，则yield必须发送所有内容。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

Answer 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

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

Answer 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

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

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)

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

Answer 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

Answer 8

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

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