generator.throw() 的作用是什么？

Question

PEP 342（通过增强生成器实现的协程）为生成器对象添加了一个 throw() 方法，它允许调用者在生成器内部抛出异常（就像由 yield 表达式抛出一样）。

我想知道这个特性的使用场景是什么。

- NikiC

---

我目前正在PHP中开发一个生成器/协程实现，我在思考是否应该包含throw()功能。 - NikiC

你想要生成器还是协程？虽然 Python 把它们混为一谈，并且你可以从后者构建前者，但它们是不同的（就像在完全不同的领域里）。 - user395760

除其他外，这还允许实现@contextmanager装饰器。 - Alexey

Answer 1

假设我使用生成器来处理添加信息到数据库的操作；我用它来存储从网络接收的信息，并通过使用生成器，只有在实际接收到数据时才能高效地执行此操作，并在其他情况下执行其他操作。

因此，我的生成器首先打开数据库连接，每次你发送一些东西时，它都会添加一行：

def add_to_database(connection_string):
    db = mydatabaselibrary.connect(connection_string)
    cursor = db.cursor()
    while True:
        row = yield
        cursor.execute('INSERT INTO mytable VALUES(?, ?, ?)', row)

这样做没问题。每次我调用.send()，它都会插入一行数据。

但如果我的数据库是事务性的呢？我如何向生成器发出信号，在何时将数据提交到数据库？在何时中止事务？此外，它一直占用着与数据库的开放连接，也许有时我想关闭该连接以回收资源。

这就是.throw()方法的用处所在；通过.throw()，我可以在该方法中引发异常来表示某些情况：

def add_to_database(connection_string):
    db = mydatabaselibrary.connect(connection_string)
    cursor = db.cursor()
    try:
        while True:
            try:
                row = yield
                cursor.execute('INSERT INTO mytable VALUES(?, ?, ?)', row)
            except CommitException:
                cursor.execute('COMMIT')
            except AbortException:
                cursor.execute('ABORT')
    finally:
        cursor.execute('ABORT')
        db.close()

.close() 方法在生成器上基本上做了相同的事情。它使用 GeneratorExit 异常和 .throw() 来关闭正在运行的生成器。

所有这些是协程工作方式的重要基础；协程本质上是生成器，加上一些额外的语法使编写协程更加容易和清晰。但在幕后，它们仍然建立在相同的 yield 和 send 上。当您并行运行多个协程时，如果其中一个协程失败，您需要一种方法来干净地退出这些协程，这只是一个例子。

- Martijn Pieters

---

谢谢您的回答。这绝对是一个有趣的用例。但我想知道这是否可以被归类为异常滥用。提交和中止不是异常情况，而是通常行为的一部分。因此，在这里，异常基本上被用作改变控制流的手段。 - NikiC

@NikiC 您的观点在同步编程方面是正确的，但您需要将其视为异步编程世界中的一部分。想象一下上面的 try 块要大得多（称调用 try 中的代码为一般用例），甚至加入几个 yield 语句，以便生成器在其一般用例期间进入和退出。.throw() 方法允许我们“跳出”以处理特殊异常。如果您熟悉中断处理程序，可以将其视为类似于那样。这样，无论在用例的哪个位置，我们都可以中断流程以执行特殊（如果不是关键的）操作。 - Paul Seeb

使用异常进行控制流并没有错。 - Marcin

Python经常使用异常来进行控制流程，例如之前提到的GeneratorExit异常。虽然像C++和Java这样的语言鼓励人们将异常的使用限制在真正的例外情况下，但Python更多地使用它们——但通常是在定义好的接口上使用。 - Andrew Aylett

我假设throw()或close()发生在cursor.execute()中？这将导致ValueError：生成器未运行，而不管您放入什么错误，都会导致Generator Exit。为什么不直接引发错误呢？ - astralwolf

Answer 2

在我看来，throw()方法有许多使用的原因。

1. 对称性：没有强烈的理由要求异常情况只能在调用者中处理而不能在生成器函数中处理。（假设一个从数据库读取值的生成器返回了错误值，并且只有调用者知道该值是错误的。使用throw()方法，调用者可以向生成器发出信号，表示存在必须更正的异常情况。）如果生成器可以引发由调用者拦截的异常，则反过来也应该是可能的。

2. 灵活性：生成器函数可能具有多个yield语句，调用者可能不知道生成器的内部状态。通过抛出异常，可以将生成器“重置”到已知状态，或者实现更复杂的流程控制，这样仅仅使用next()、send()、close()将会更加麻烦。

以下是重置内部状态的示例：

def gen():
    try:
        yield 10
        print("State1")
        yield 20
        print("State2")
        yield 30
        print("State3")
    
   except:
        #Reset back to State1!
        yield gen()

g = gen()
print(next(g))
print(next(g))
g = g.throw(ValueError) #state of g has been reset
print(next(g))

>>10
>>State1
>>20
>>10

询问使用情况可能会产生误导：对于每个使用情况，您都可以提供一个反例而无需使用 throw() 方法，并且讨论将继续下去...

- Stefano M

---

能否举个例子，说明在什么情况下使用 throw 可以将生成器重置到已知状态？ - astralwolf

Answer 3

一个使用案例是在异常发生时，在堆栈跟踪中包含有关生成器内部状态的信息 -- 否则调用者将看不到这些信息。

例如，假设我们有一个生成器，如下所示，其中我们想要的内部状态是生成器的当前索引号：

def gen_items():
    for i, item in enumerate(["", "foo", "", "foo", "bad"]):
        if not item:
            continue
        try:
            yield item
        except Exception:
            raise Exception("error during index: %d" % i)

以下代码不足以触发额外的异常处理：

# Stack trace includes only: "ValueError: bad value"
for item in gen_items():
    if item == "bad":
        raise ValueError("bad value")

然而，以下代码确实提供了内部状态：

# Stack trace also includes: "Exception: error during index: 4"
gen = item_generator()
for item in gen:
    if item == "bad":
        gen.throw(ValueError, "bad value")

- cjerdonek

Answer 4

这个“答案”更像是一道小知识点。我们可以利用生成器的throw()在lambda中抛出异常，这样就可以使用不支持raise语句的lambda了。

foo = lambda: (_ for _ in ()).throw(Exception('foobar'))

引用自https://dev59.com/o2sy5IYBdhLWcg3w0RXT#8294654

在IT技术中，“MVC”代表“模型-视图-控制器”。它是一种软件设计模式，旨在将应用程序的不同方面分离开来。模型表示应用程序中的数据和业务逻辑，视图表示用户界面，而控制器则处理输入并根据模型和视图执行相应操作。使用MVC可以增强代码的可维护性和重用性，提高开发效率，并支持团队协作。

- RayLuo

Answer 5

我正在使用它来编写可重用的库代码，可以同时具有同步和异步代码路径。它被简化为类似于这样的形式：

from abc import ABCMeta, abstractmethod
from typing import Generator

class Helper( metaclass = ABCMeta ):
    @abstractmethod
    def help( self, con: DatabaseConnection ) -> None:
        raise NotImplementedError
    @abstractmethod
    async def ahelp( self, con: AsyncDataConnection ) -> None:
        raise NotImplementedError

class HelperSelect( Helper ):
    ' logic here to execute a select query against the database '
    rows: list[dict[str,Any]] # help() and ahelp() write their results here

    def help( self, con: DatabaseConnection ) -> None:
        assert False, 'TODO FIXME write database execution logic here'

    async def ahelp( self, con: AsyncDataConnection ) -> None:
        assert False, 'TODO FIXME write database execution logic here'

def _application_logic() -> Generator[Helper,None,int]:
    sql = 'select * from foo'
    helper = HelperSelect( sql )
    yield helper
    # do something with helper.rows
    return 0

def do_something( con: DatabaseConnection ):
    gen = _application_logic()
    try:
        while True:
            helper = next( gen )
            try:
                helper.help( con )
            except Exception as e:
                gen.throw( e )
    except StopIteration as e:
        return e.value

async def ado_something( con: AsyncDatabaseConnection ):
    gen = _application_logic()
    try:
        while True:
            helper = next( gen )
            try:
                await helper.ahelp( con )
            except Exception as e:
                gen.throw( e )
    except StopIteration as e:
        return e.value

如果不使用gen.throw，堆栈跟踪就不会显示异常发生的逻辑位置，这可能非常令人沮丧。使用上面示例中的gen.throw（）可以解决这个问题。

Helper类的原因是，除了数据库查询之外，逻辑可能需要请求半打不同类型的事情，需要异步执行。

我采用了伪代码，并构建了一个可以实际运行并查看差异的版本：

from abc import ABCMeta, abstractmethod
import logging
from typing import Any, Generator

class Helper( metaclass = ABCMeta ):
    @abstractmethod
    def help( self ) -> None:
        raise NotImplementedError

class HelperBoom( Helper ):
    def help( self ) -> None:
        assert False

def _application_logic() -> Generator[Helper,None,int]:
    helper = HelperBoom()
    yield helper
    return 0

def do_something1():
    gen = _application_logic()
    try:
        while True:
            helper = next( gen )
            helper.help()
    except StopIteration as e:
        return e.value

def do_something2():
    gen = _application_logic()
    try:
        while True:
            helper = next( gen )
            try:
                helper.help()
            except Exception as e:
                gen.throw( e )
    except StopIteration as e:
        return e.value

try:
    do_something1()
except Exception:
    logging.exception( 'do_something1 failed:' )
try:
    do_something2()
except Exception:
    logging.exception( 'do_something2 failed:' )

这是输出结果，注意do_something1的堆栈跟踪中缺少_application_logic行，但do_something2的堆栈跟踪中有该条目。

ERROR:root:do_something1 failed:
Traceback (most recent call last):
  File "C:\cvs\itas\incpy\test_helper_exceptions.py", line 35, in <module>
    do_something1()
  File "C:\cvs\itas\incpy\test_helper_exceptions.py", line 23, in do_something1
    helper.help()
  File "C:\cvs\itas\incpy\test_helper_exceptions.py", line 12, in help
    assert False
AssertionError
ERROR:root:do_something2 failed:
Traceback (most recent call last):
  File "C:\cvs\itas\incpy\test_helper_exceptions.py", line 39, in <module>
    do_something2()
  File "C:\cvs\itas\incpy\test_helper_exceptions.py", line 32, in do_something2
    gen.throw( e )
  File "C:\cvs\itas\incpy\test_helper_exceptions.py", line 16, in _application_logic
    yield helper
  File "C:\cvs\itas\incpy\test_helper_exceptions.py", line 30, in do_something2
    helper.help()
  File "C:\cvs\itas\incpy\test_helper_exceptions.py", line 12, in help
    assert False
AssertionError

- royce3