私はジェネレーターとそれを消費する関数を持っています：
def read():
while something():
yield something_else()
def process():
for item in read():
do stuff
ジェネレーターが例外をスローした場合、コンシューマー関数でそれを処理してから、イテレーターが使い果たされるまでイテレーターを消費し続けたいと思います。ジェネレーターに例外処理コードを含めたくないことに注意してください。
私は次のようなことを考えました：
reader = read()
while True:
try:
item = next(reader)
except StopIteration:
break
except Exception as e:
log error
continue
do_stuff(item)
しかし、これは私にはかなり厄介に見えます。
ジェネレータが例外をスローすると、ジェネレータは終了します。生成されたアイテムを継続して消費することはできません。
例：
>>> def f():
... yield 1
... raise Exception
... yield 2
...
>>> g = f()
>>> next(g)
1
>>> next(g)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "<stdin>", line 3, in f
Exception
>>> next(g)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
StopIteration
ジェネレーターコードを制御する場合は、ジェネレーター内で例外を処理できます。そうでない場合は、例外の発生を回避するようにしてください。
これも、正しく/エレガントに処理できるかどうかわからないことです。
私がやっていることはyield、Exceptionジェネレーターから、そしてそれをどこか別の場所に上げることです。好き：
class myException(Exception):
def __init__(self, ...)
...
def g():
...
if everything_is_ok:
yield result
else:
yield myException(...)
my_gen = g()
while True:
try:
n = next(my_gen)
if isinstance(n, myException):
raise n
except StopIteration:
break
except myException as e:
# Deal with exception, log, print, continue, break etc
else:
# Consume n
このようにして、例外を発生させずに引き継ぐことができます。これにより、ジェネレーター関数が停止します。isinstance主な欠点は、各反復で生成された結果を確認する必要があることです。さまざまなタイプの結果を生成できるジェネレーターは好きではありませんが、最後の手段として使用します。
私はこの問題を数回解決する必要があり、他の人が何をしたかを調べた後、この質問に出くわしました。
1つのオプション（少しリファクタリングが必要になりthrowます）は、ジェネレーターの例外（別のエラー処理ジェネレーター）ではなく、例外raiseです。これがどのように見えるかです：
def read(handler):
# the handler argument fixes errors/problems separately
while something():
try:
yield something_else()
except Exception as e:
handler.throw(e)
handler.close()
def err_handler():
# a generator for processing errors
while True:
try:
yield
except Exception1:
handle_exc1()
except Exception2:
handle_exc2()
except Exception3:
handle_exc3()
except Exception:
raise
def process():
handler = err_handler()
handler.send(None) # initialize error handler
for item in read(handler):
do stuff
これが常に最良の解決策になるとは限りませんが、それは確かにオプションです。
デコレータを使用すると、すべてを少しだけ良くすることができます。
class MyError(Exception):
pass
def handled(handler):
"""
A decorator that applies error handling to a generator.
The handler argument received errors to be handled.
Example usage:
@handled(err_handler())
def gen_function():
yield the_things()
"""
def handled_inner(gen_f):
def wrapper(*args, **kwargs):
g = gen_f(*args, **kwargs)
while True:
try:
g_next = next(g)
except StopIteration:
break
if isinstance(g_next, Exception):
handler.throw(g_next)
else:
yield g_next
return wrapper
handler.send(None) # initialize handler
return handled_inner
def my_err_handler():
while True:
try:
yield
except MyError:
print("error handled")
# all other errors will bubble up here
@handled(my_err_handler())
def read():
i = 0
while i<10:
try:
yield i
i += 1
if i == 3:
raise MyError()
except Exception as e:
# prevent the generator from closing after an Exception
yield e
def process():
for item in read():
print(item)
if __name__=="__main__":
process()
出力：
0
1
2
error handled
3
4
5
6
7
8
9
ただし、これの欠点は、Exceptionエラーを生成する可能性のあるジェネレーター内に一般的な処理を配置する必要があることです。ジェネレータで例外を発生させるとクローズされるため、これを回避することはできません。
yield raiseエラーが発生した後、可能であればジェネレーターが実行を継続できるようにする、ある種のステートメントがあると便利です。次に、次のようなコードを記述できます。
@handled(my_err_handler())
def read():
i = 0
while i<10:
yield i
i += 1
if i == 3:
yield raise MyError()
...そしてhandler()デコレータは次のようになります：
def handled(handler):
def handled_inner(gen_f):
def wrapper(*args, **kwargs):
g = gen_f(*args, **kwargs)
while True:
try:
g_next = next(g)
except StopIteration:
break
except Exception as e:
handler.throw(e)
else:
yield g_next
return wrapper
handler.send(None) # initialize handler
return handled_inner
Python 3.3以降、元のジェネレーターから例外をキャッチするためのコードは非常に単純になります。
from types import GeneratorType
def gen_decorator(func):
def gen_wrapper(generator):
try:
yield from generator # I mean this line!
except Exception:
print('catched in gen_decorator while iterating!'.upper())
raise
def wrapper():
try:
result = func()
if isinstance(result, GeneratorType):
result = gen_wrapper(result)
return result
except Exception:
print('catched in gen_decorator while initialization!'.upper())
raise
return wrapper
そして使用例：
@gen_decorator
def gen():
x = 0
while True:
x += 1
if x == 5:
raise RuntimeError('error!')
yield x
if __name__ == '__main__':
try:
for i in gen():
print(i)
if i >= 10:
print('lets stop!')
break
except Exception:
print('catched in main!'.upper())
raise