Зачем генератору метод throw?

Метод throw() для генераторов

Метод throw() позволяет инъектировать исключение в генератор без внешнего контекста. Это мощный инструмент для обработки ошибок в асинхронном коде.

Базовый пример

def my_generator():
    try:
        print("Starting")
        yield 1
        print("After first yield")
        yield 2
        print("After second yield")
        yield 3
    except ValueError as e:
        print(f"Caught exception: {e}")
        yield "error handled"

gen = my_generator()
print(next(gen))           # Выведет: Starting, результат: 1
print(next(gen))           # Выведет: After first yield, результат: 2

# Инъектирую исключение
result = gen.throw(ValueError("Something went wrong"))
print(result)              # error handled

try:
    next(gen)              # Больше значений нет
except StopIteration:
    print("Generator finished")

Как это работает:

1. Генератор приостановлен на yield 2
2. Вызов throw() переводит исключение в точку yield 2
3. try/except ловит это исключение
4. Генератор продолжает выполнение (может выдать новое значение)

Где применяется

1. Context managers для генераторов

Это основное использование throw() — для управления ресурсами:

from contextlib import contextmanager

@contextmanager
def database_connection(host: str):
    print(f"Connecting to {host}")
    conn = None
    try:
        conn = connect_to_db(host)  # Имитация
        yield conn  # Возвращаю соединение
    except Exception as e:
        print(f"Error occurred: {e}")
        if conn:
            conn.rollback()
    finally:
        print(f"Closing connection to {host}")
        if conn:
            conn.close()

# Использование
with database_connection("localhost") as conn:
    try:
        # Может произойти ошибка
        result = conn.execute("SELECT * FROM users")
    except ConnectionError:
        # Внутри contextmanager будет вызван throw()
        pass

Внутри контекстного менеджера при исключении вызывается gen.throw(), который позволяет:

1. Обработать ошибку
2. Очистить ресурсы
3. Решить: пробросить исключение дальше или нет

2. asyncio и корутины

В asyncio throw() используется для отмены (cancellation) задач:

import asyncio

async def long_running_task():
    try:
        print("Task started")
        for i in range(10):
            await asyncio.sleep(1)
            print(f"Working: {i}")
    except asyncio.CancelledError:
        print("Task was cancelled")
        # Очистка ресурсов
        await cleanup_resources()
        raise

async def main():
    task = asyncio.create_task(long_running_task())
    await asyncio.sleep(3)
    task.cancel()  # Внутри вызовет throw(asyncio.CancelledError)
    try:
        await task
    except asyncio.CancelledError:
        print("Task cancellation confirmed")

asyncio.run(main())
# Вывод:
# Task started
# Working: 0
# Working: 1
# Working: 2
# Task was cancelled
# Task cancellation confirmed

3. Обработка ошибок в middleware

Можно использовать для управления потоком обработки в pipeline:

def error_handler_middleware(generator):
    """
    Middleware, который ловит ошибки и позволяет генератору их обработать
    """
    while True:
        try:
            print("Middleware: sending data to generator")
            data = yield
            result = generator.send(data)
            yield result
        except Exception as e:
            print(f"Middleware: caught {type(e).__name__}")
            # Передаю ошибку в генератор
            result = generator.throw(type(e), e)
            yield result

def my_processor():
    try:
        print("Processor: ready")
        data = yield
        print(f"Processor: got {data}")
        if data < 0:
            raise ValueError("Negative value not allowed")
        yield data * 2
    except ValueError as e:
        print(f"Processor: handled error: {e}")
        yield -1  # Return error code

# Использование
proc = my_processor()
handler = error_handler_middleware(proc)

next(handler)
print("Result:", next(handler))  # Result: 10

Когда throw() действительно нужен

1. Graceful shutdown

import asyncio
import signal

class TaskManager:
    def __init__(self):
        self.generators = []
        signal.signal(signal.SIGTERM, self._handle_shutdown)
    
    def _handle_shutdown(self, signum, frame):
        print("Shutdown signal received")
        for gen in self.generators:
            try:
                gen.throw(KeyboardInterrupt())
            except (StopIteration, KeyboardInterrupt):
                pass

task_manager = TaskManager()

def worker():
    try:
        while True:
            print("Worker running")
            yield
    except KeyboardInterrupt:
        print("Worker: graceful shutdown")
        # Cleanup
        yield

2. Timeout обработка

import threading

def timeout_wrapper(generator, timeout_seconds):
    """
    Если генератор не вернёт значение за timeout — инъектирую TimeoutError
    """
    def timeout_handler():
        try:
            generator.throw(TimeoutError("Generator timeout"))
        except (StopIteration, TimeoutError):
            pass
    
    timer = threading.Timer(timeout_seconds, timeout_handler)
    timer.start()
    
    try:
        result = yield from generator
    finally:
        timer.cancel()
    
    return result

def slow_generator():
    try:
        print("Starting slow operation")
        yield
        import time
        time.sleep(10)  # Медленная операция
        yield "done"
    except TimeoutError:
        print("Operation timed out")
        yield "error"

API метода throw()

def generator():
    try:
        yield 1
    except ValueError:
        yield 2

gen = generator()
next(gen)

# Способ 1: throw(exception_instance)
gen.throw(ValueError("message"))

# Способ 2: throw(exception_class, value)
gen.throw(ValueError, ValueError("message"))

# Способ 3: throw(exception_class, value, traceback)
import sys
tb = sys.exc_info()[2]
gen.throw(ValueError, ValueError("message"), tb)

Важные детали

def gen():
    try:
        yield 1
    except ValueError:
        pass

g = gen()
next(g)

# Если генератор не обработает исключение — оно пробросится дальше
try:
    g.throw(ValueError)
except ValueError:
    print("Exception not handled by generator")

# Если генератор обработает — можно получить следующее значение
def gen2():
    try:
        yield 1
    except ValueError:
        yield 2

g2 = gen2()
next(g2)
result = g2.throw(ValueError)  # Вернёт 2
print(result)

close() vs throw()

def gen():
    try:
        yield 1
        yield 2
    finally:
        print("Cleanup")

g = gen()
next(g)

# close() = throw(GeneratorExit)
g.close()  # Выведет: Cleanup

# Эквивалентно:
g2 = gen()
next(g2)
g2.throw(GeneratorExit)  # Тоже выведет: Cleanup

Вывод

throw() используется для:

• ✓ Контекстных менеджеров (обработка ошибок при выходе)
• ✓ Отмены задач в asyncio (CancelledError)
• ✓ Управления ошибками в pipeline'ах
• ✓ Timeout обработки
• ✓ Graceful shutdown

Не используй если:

• ❌ Можешь обработать ошибку в вызывающем коде
• ❌ Нужна простая последовательная обработка (используй try/except)

Это advanced feature, которая нужна в специфичных случаях (фреймворки,低-уровневый код).