Python における yield 文の管理

はじめに

この実験では、Python の yield 文で何が起こるかを管理する方法を学びます。これらの文に関連する操作と動作を効果的に処理する方法を理解することができます。

さらに、ジェネレータのライフタイムとジェネレータ内の例外処理について学びます。この学習過程では、follow.py と cofollow.py のファイルを変更します。

ジェネレータのライフタイムとクロージャの理解

このステップでは、Python のジェネレータのライフタイムを調べ、適切にクローズする方法を学びます。Python のジェネレータは、特別な種類のイテレータで、一度にすべての値を計算してメモリに格納するのではなく、必要に応じて値のシーケンスを生成することができます。これは、大規模なデータセットや無限シーケンスを扱う場合に非常に便利です。

follow() ジェネレータとは？

まず、プロジェクトディレクトリ内の follow.py ファイルを見てみましょう。このファイルには、follow() という名前のジェネレータ関数が含まれています。ジェネレータ関数は通常の関数のように定義されますが、return キーワードの代わりに yield を使用します。ジェネレータ関数が呼び出されると、ジェネレータオブジェクトが返され、これを反復処理することで、生成される値を取得できます。

follow() ジェネレータ関数は、ファイルから行を継続的に読み取り、読み取った各行を生成します。これは、新しい行を継続的に監視する Unix の tail -f コマンドに似ています。

WebIDE エディタで follow.py ファイルを開きます。

import os
import time

def follow(filename):
    with open(filename,'r') as f:
        f.seek(0,os.SEEK_END)
        while True:
            line = f.readline()
            if line == '':
                time.sleep(0.1)    ## Sleep briefly to avoid busy wait
                continue
            yield line

このコードでは、with open(filename, 'r') as f 文がファイルを読み取りモードで開き、ブロックを抜けるときに適切に閉じることを保証します。f.seek(0, os.SEEK_END) 行は、ファイルポインタをファイルの末尾に移動させるため、ジェネレータは末尾から読み取りを開始します。while True ループは、ファイルから行を継続的に読み取ります。行が空の場合、まだ新しい行がないことを意味するので、プログラムは 0.1 秒間スリープしてビジーウェイトを避け、次の反復に進みます。行が空でない場合、その行が生成されます。

このジェネレータは無限ループで実行されます。これにより、重要な質問が生じます。ジェネレータの使用を停止する場合、または早期に終了させたい場合、何が起こるのでしょうか？

クロージャを処理するためのジェネレータの修正

follow.py の follow() 関数を修正して、ジェネレータが適切にクローズされた場合を処理する必要があります。これを行うには、GeneratorExit 例外をキャッチする try-except ブロックを追加します。GeneratorExit 例外は、ガベージコレクションまたは close() メソッドの呼び出しによってジェネレータがクローズされたときに発生します。

import os
import time

def follow(filename):
    try:
        with open(filename,'r') as f:
            f.seek(0,os.SEEK_END)
            while True:
                line = f.readline()
                if line == '':
                    time.sleep(0.1)    ## Sleep briefly to avoid busy wait
                    continue
                yield line
    except GeneratorExit:
        print('Following Done')

この修正されたコードでは、try ブロックにジェネレータの主要なロジックが含まれています。GeneratorExit 例外が発生した場合、except ブロックがそれをキャッチし、メッセージ 'Following Done' を出力します。これは、ジェネレータがクローズされたときにクリーンアップアクションを実行する簡単な方法です。

これらの変更を加えた後、ファイルを保存します。

ジェネレータのクロージャの実験

では、ジェネレータがガベージコレクションされたり、明示的にクローズされたりしたときの動作を確認するために、いくつかの実験を行いましょう。

ターミナルを開き、Python インタープリタを起動します。

cd ~/project
python3

実験 1: 実行中のジェネレータのガベージコレクション

>>> from follow import follow
>>> ## Experiment: Garbage collection of a running generator
>>> f = follow('stocklog.csv')
>>> next(f)
'"MO",70.29,"6/11/2007","09:30.09",-0.01,70.25,70.30,70.29,365314\n'
>>> del f  ## Delete the generator object
Following Done  ## This message appears because of our GeneratorExit handler

この実験では、まず follow.py ファイルから follow 関数をインポートします。次に、follow('stocklog.csv') を呼び出してジェネレータオブジェクト f を作成します。next() 関数を使用して、ジェネレータから次の行を取得します。最後に、del 文を使用してジェネレータオブジェクトを削除します。ジェネレータオブジェクトが削除されると、自動的にクローズされ、GeneratorExit 例外ハンドラがトリガーされ、メッセージ 'Following Done' が出力されます。

実験 2: ジェネレータの明示的なクローズ

>>> f = follow('stocklog.csv')
>>> for line in f:
...     print(line, end='')
...     if 'IBM' in line:
...         f.close()  ## Explicitly close the generator
...
"MO",70.29,"6/11/2007","09:30.09",-0.01,70.25,70.30,70.29,365314
"VZ",42.91,"6/11/2007","09:34.28",-0.16,42.95,42.91,42.78,210151
"HPQ",45.76,"6/11/2007","09:34.29",0.06,45.80,45.76,45.59,257169
"GM",31.45,"6/11/2007","09:34.31",0.45,31.00,31.50,31.45,582429
"IBM",102.86,"6/11/2007","09:34.44",-0.21,102.87,102.86,102.77,147550
Following Done
>>> for line in f:
...     print(line, end='')  ## No output: generator is closed
...

この実験では、新しいジェネレータオブジェクト f を作成し、for ループを使用して反復処理します。ループ内では、各行を出力し、行に文字列 'IBM' が含まれているかどうかを確認します。含まれている場合、ジェネレータの close() メソッドを呼び出して明示的にクローズします。ジェネレータがクローズされると、GeneratorExit 例外が発生し、例外ハンドラがメッセージ 'Following Done' を出力します。ジェネレータがクローズされた後、再度反復処理を試みると、ジェネレータがもはやアクティブでないため、出力はありません。

実験 3: ジェネレータのループからの脱出と再開

>>> f = follow('stocklog.csv')
>>> for line in f:
...     print(line, end='')
...     if 'IBM' in line:
...         break  ## Break out of the loop, but don't close the generator
...
"MO",70.29,"6/11/2007","09:30.09",-0.01,70.25,70.30,70.29,365314
"VZ",42.91,"6/11/2007","09:34.28",-0.16,42.95,42.91,42.78,210151
"HPQ",45.76,"6/11/2007","09:34.29",0.06,45.80,45.76,45.59,257169
"GM",31.45,"6/11/2007","09:34.31",0.45,31.00,31.50,31.45,582429
"IBM",102.86,"6/11/2007","09:34.44",-0.21,102.87,102.86,102.77,147550
>>> ## Resume iteration - the generator is still active
>>> for line in f:
...     print(line, end='')
...     if 'IBM' in line:
...         break
...
"CAT",78.36,"6/11/2007","09:37.19",-0.16,78.32,78.36,77.99,237714
"VZ",42.99,"6/11/2007","09:37.20",-0.08,42.95,42.99,42.78,268459
"IBM",102.91,"6/11/2007","09:37.31",-0.16,102.87,102.91,102.77,190859
>>> del f  ## Clean up
Following Done

この実験では、ジェネレータオブジェクト f を作成し、for ループを使用して反復処理します。ループ内では、各行を出力し、行に文字列 'IBM' が含まれているかどうかを確認します。含まれている場合、break 文を使用してループから脱出します。ループから脱出してもジェネレータはクローズされないため、ジェネレータは依然としてアクティブです。その後、同じジェネレータオブジェクトに対して新しい for ループを開始することで、反復処理を再開できます。最後に、ジェネレータオブジェクトを削除してクリーンアップし、GeneratorExit 例外ハンドラがトリガーされます。

要点

1. ジェネレータがクローズされると（ガベージコレクションまたは close() の呼び出しによる）、ジェネレータ内で GeneratorExit 例外が発生します。
2. この例外をキャッチすることで、ジェネレータがクローズされたときにクリーンアップアクションを実行できます。
3. ジェネレータの反復処理から脱出する（break を使用）と、ジェネレータはクローズされず、後で再開できます。

exit() と入力するか、Ctrl+D を押して Python インタープリタを終了します。

ジェネレータでの例外処理

このステップでは、ジェネレータとコルーチンで例外を処理する方法を学びます。まずは、例外とは何かを理解しましょう。例外とは、プログラムの実行中に発生し、プログラムの命令の通常の流れを中断するイベントです。Python では、throw() メソッドを使用して、ジェネレータとコルーチンで例外を処理することができます。

コルーチンの理解

コルーチンは、特殊な種類のジェネレータです。主に値を生成する通常のジェネレータとは異なり、コルーチンは値を消費することも（send() メソッドを使用）、値を生成することもできます。cofollow.py ファイルには、コルーチンの簡単な実装があります。

WebIDE エディタで cofollow.py ファイルを開きましょう。中身のコードは次の通りです。

def consumer(func):
    def start(*args,**kwargs):
        c = func(*args,**kwargs)
        next(c)
        return c
    return start

@consumer
def printer():
    while True:
        item = yield
        print(item)

では、このコードを分解してみましょう。consumer はデコレータです。デコレータは、別の関数を引数として受け取り、それにいくつかの機能を追加してから、修正された関数を返す関数です。この場合、consumer デコレータは自動的にジェネレータを最初の yield 文まで進めます。これは、ジェネレータが値を受け取る準備をするために重要です。

printer() コルーチンは、@consumer デコレータで定義されています。printer() 関数の中には、無限の while ループがあります。item = yield 文がポイントです。これはコルーチンの実行を一時停止し、値を受け取るのを待ちます。コルーチンに値が送られると、実行が再開され、受け取った値が出力されます。

コルーチンに例外処理を追加する

では、printer() コルーチンを修正して例外を処理するようにしましょう。cofollow.py の printer() 関数を次のように更新します。

@consumer
def printer():
    while True:
        try:
            item = yield
            print(item)
        except Exception as e:
            print('ERROR: %r' % e)

try ブロックには、例外を引き起こす可能性のあるコードが含まれています。この場合、値を受け取って出力するコードです。try ブロックで例外が発生すると、実行は except ブロックにジャンプします。except ブロックは例外をキャッチし、エラーメッセージを出力します。これらの変更を加えた後、ファイルを保存します。

コルーチンでの例外処理の実験

では、コルーチンに例外を投げる実験を始めましょう。ターミナルを開き、次のコマンドを使用して Python インタープリタを起動します。

cd ~/project
python3

実験 1: 基本的なコルーチンの使用

>>> from cofollow import printer
>>> p = printer()
>>> p.send('hello')  ## Send a value to the coroutine
hello
>>> p.send(42)  ## Send another value
42

ここでは、まず cofollow モジュールから printer コルーチンをインポートします。次に、printer コルーチンのインスタンス p を作成します。send() メソッドを使用して、コルーチンに値を送ります。見ての通り、コルーチンは送られた値を問題なく処理します。

実験 2: コルーチンに例外を投げる

>>> p.throw(ValueError('It failed'))  ## Throw an exception into the coroutine
ERROR: ValueError('It failed')

この実験では、throw() メソッドを使用して、ValueError 例外をコルーチンに注入します。printer() コルーチンの try-except ブロックが例外をキャッチし、エラーメッセージを出力します。これは、例外処理が期待通りに機能していることを示しています。

実験 3: コルーチンに実際の例外を投げる

>>> try:
...     int('n/a')  ## This will raise a ValueError
... except ValueError as e:
...     p.throw(e)  ## Throw the caught exception into the coroutine
...
ERROR: ValueError("invalid literal for int() with base 10: 'n/a'")

ここでは、まず文字列 'n/a' を整数に変換しようとしますが、これは ValueError を引き起こします。この例外をキャッチしてから、throw() メソッドを使用してコルーチンに渡します。コルーチンは例外をキャッチし、エラーメッセージを出力します。

実験 4: コルーチンが引き続き実行されることを確認する

>>> p.send('still working')  ## The coroutine continues to run after handling exceptions
still working

例外を処理した後、send() メソッドを使用してコルーチンに別の値を送ります。コルーチンはまだアクティブで、新しい値を処理することができます。これは、コルーチンがエラーに遭遇した後も引き続き実行できることを示しています。

要点

1. ジェネレータとコルーチンは、yield 文の位置で例外を処理することができます。これは、コルーチンが値を待っているときや処理しているときに発生するエラーをキャッチして処理できることを意味します。
2. throw() メソッドを使用すると、ジェネレータまたはコルーチンに例外を注入することができます。これは、テストやコルーチンの外で発生したエラーを処理するのに役立ちます。
3. ジェネレータで適切に例外を処理することで、エラーが発生しても引き続き実行できる堅牢でエラー耐性のあるジェネレータを作成することができます。これにより、コードがより信頼性が高く、保守が容易になります。

Python インタープリタを終了するには、exit() と入力するか、Ctrl+D を押します。

ジェネレータ管理の実用的なアプリケーション

このステップでは、ジェネレータの管理とジェネレータ内での例外処理について学んだ概念を、実世界のシナリオにどのように適用するかを探ります。これらの実用的なアプリケーションを理解することで、より堅牢で効率的な Python コードを書くことができます。

堅牢なファイル監視システムの作成

より信頼性の高いファイル監視システムを構築しましょう。このシステムは、タイムアウトやユーザーからの停止要求など、さまざまな状況を処理できるようになります。

まず、WebIDE エディタを開き、robust_follow.py という名前の新しいファイルを作成します。このファイルに書くコードは次の通りです。

import os
import time
import signal

class TimeoutError(Exception):
    pass

def timeout_handler(signum, frame):
    raise TimeoutError("Operation timed out")

def follow(filename, timeout=None):
    """
    A generator that yields new lines in a file.
    With timeout handling and proper cleanup.
    """
    try:
        ## Set up timeout if specified
        if timeout:
            signal.signal(signal.SIGALRM, timeout_handler)
            signal.alarm(timeout)

        with open(filename, 'r') as f:
            f.seek(0, os.SEEK_END)
            while True:
                line = f.readline()
                if line == '':
                    ## No new data, wait briefly
                    time.sleep(0.1)
                    continue
                yield line
    except TimeoutError:
        print(f"Following timed out after {timeout} seconds")
    except GeneratorExit:
        print("Following stopped by request")
    finally:
        ## Clean up timeout alarm if it was set
        if timeout:
            signal.alarm(0)
        print("Follow generator cleanup complete")

このコードでは、まずカスタムの TimeoutError クラスを定義しています。timeout_handler 関数は、タイムアウトが発生したときにこのエラーを発生させるために使用されます。follow 関数は、ファイルを読み取り、新しい行を生成するジェネレータです。タイムアウトが指定されている場合、signal モジュールを使用してアラームを設定します。ファイルに新しいデータがない場合は、しばらく待ってから再度試みます。try - except - finally ブロックは、さまざまな例外を処理し、適切なクリーンアップを行うために使用されます。

コードを書いたら、ファイルを保存します。

堅牢なファイル監視システムの実験

では、改良したファイル監視システムをテストしましょう。ターミナルを開き、次のコマンドで Python インタープリタを起動します。

cd ~/project
python3

実験 1: 基本的な使用方法

Python インタープリタで、follow ジェネレータの基本的な機能をテストします。実行するコードは次の通りです。

>>> from robust_follow import follow
>>> f = follow('stocklog.csv')
>>> for i, line in enumerate(f):
...     print(f"Line {i+1}: {line.strip()}")
...     if i >= 2:  ## Just read a few lines for the example
...         break
...
Line 1: "MO",70.29,"6/11/2007","09:30.09",-0.01,70.25,70.30,70.29,365314
Line 2: "VZ",42.91,"6/11/2007","09:34.28",-0.16,42.95,42.91,42.78,210151
Line 3: "HPQ",45.76,"6/11/2007","09:34.29",0.06,45.80,45.76,45.59,257169

ここでは、robust_follow.py ファイルから follow 関数をインポートしています。そして、stocklog.csv ファイルを監視するジェネレータオブジェクト f を作成します。for ループを使用して、ジェネレータが生成する行を反復処理し、最初の 3 行を出力します。

実験 2: タイムアウトの使用

タイムアウト機能がどのように動作するかを見てみましょう。Python インタープリタで次のコードを実行します。

>>> ## Create a generator that will time out after 3 seconds
>>> f = follow('stocklog.csv', timeout=3)
>>> for line in f:
...     print(line.strip())
...     time.sleep(1)  ## Process each line slowly
...
"MO",70.29,"6/11/2007","09:30.09",-0.01,70.25,70.30,70.29,365314
"VZ",42.91,"6/11/2007","09:34.28",-0.16,42.95,42.91,42.78,210151
"HPQ",45.76,"6/11/2007","09:34.29",0.06,45.80,45.76,45.59,257169
Following timed out after 3 seconds
Follow generator cleanup complete

この実験では、3 秒でタイムアウトするジェネレータを作成しています。各行の処理を 1 秒間スリープすることでゆっくりと行います。約 3 秒後、ジェネレータはタイムアウト例外を発生させ、finally ブロックのクリーンアップコードが実行されます。

実験 3: 明示的なクローズ

ジェネレータが明示的なクローズをどのように処理するかをテストしましょう。次のコードを実行します。

>>> f = follow('stocklog.csv')
>>> for i, line in enumerate(f):
...     print(f"Line {i+1}: {line.strip()}")
...     if i >= 1:
...         print("Explicitly closing the generator...")
...         f.close()
...
Line 1: "MO",70.29,"6/11/2007","09:30.09",-0.01,70.25,70.30,70.29,365314
Line 2: "VZ",42.91,"6/11/2007","09:34.28",-0.16,42.95,42.91,42.78,210151
Explicitly closing the generator...
Following stopped by request
Follow generator cleanup complete

ここでは、ジェネレータを作成し、その行を反復処理し始めます。2 行を処理した後、close メソッドを使用してジェネレータを明示的にクローズします。その後、ジェネレータは GeneratorExit 例外を処理し、必要なクリーンアップを行います。

エラー処理を備えたデータ処理パイプラインの作成

次に、コルーチンを使用して簡単なデータ処理パイプラインを作成します。このパイプラインは、さまざまな段階でエラーを処理できるようになります。

WebIDE エディタを開き、pipeline.py という名前の新しいファイルを作成します。このファイルに書くコードは次の通りです。

def consumer(func):
    def start(*args,**kwargs):
        c = func(*args,**kwargs)
        next(c)
        return c
    return start

@consumer
def grep(pattern, target):
    """Filter lines containing pattern and send to target"""
    try:
        while True:
            line = yield
            if pattern in line:
                target.send(line)
    except Exception as e:
        target.throw(e)

@consumer
def printer():
    """Print received items"""
    try:
        while True:
            item = yield
            print(f"PRINTER: {item}")
    except Exception as e:
        print(f"PRINTER ERROR: {repr(e)}")

def follow_and_process(filename, pattern):
    """Follow a file and process its contents"""
    import time
    import os

    output = printer()
    filter_pipe = grep(pattern, output)

    try:
        with open(filename, 'r') as f:
            f.seek(0, os.SEEK_END)
            while True:
                line = f.readline()
                if not line:
                    time.sleep(0.1)
                    continue
                filter_pipe.send(line)
    except KeyboardInterrupt:
        print("Processing stopped by user")
    finally:
        filter_pipe.close()
        output.close()

このコードでは、consumer デコレータを使用してコルーチンを初期化しています。grep コルーチンは、特定のパターンを含む行をフィルタリングし、別のコルーチンに送信します。printer コルーチンは、受け取ったアイテムを出力します。follow_and_process 関数は、ファイルを読み取り、grep コルーチンを使用して行をフィルタリングし、printer コルーチンを使用して一致する行を出力します。また、KeyboardInterrupt 例外を処理し、適切なクリーンアップを行います。

コードを書いたら、ファイルを保存します。

データ処理パイプラインのテスト

データ処理パイプラインをテストしましょう。ターミナルで次のコマンドを実行します。

cd ~/project
python3 -c "from pipeline import follow_and_process; follow_and_process('stocklog.csv', 'IBM')"

次のような出力が表示されるはずです。

PRINTER: "IBM",102.86,"6/11/2007","09:34.44",-0.21,102.87,102.86,102.77,147550

PRINTER: "IBM",102.91,"6/11/2007","09:37.31",-0.16,102.87,102.91,102.77,190859

PRINTER: "IBM",102.95,"6/11/2007","09:39.44",-0.12,102.87,102.95,102.77,225350

この出力は、パイプラインが正しく動作しており、"IBM" パターンを含む行をフィルタリングして出力していることを示しています。

プロセスを停止するには、Ctrl+C を押します。次のメッセージが表示されるはずです。

Processing stopped by user

要点

1. ジェネレータで適切な例外処理を行うことで、エラーをうまく処理できる堅牢なシステムを作成することができます。これは、何かがうまくいかないときにプログラムが予期せずクラッシュしないことを意味します。
2. タイムアウトなどの手法を使用することで、ジェネレータが無限に実行されるのを防ぐことができます。これにより、システムリソースを管理し、プログラムが無限ループに陥るのを防ぐことができます。
3. ジェネレータとコルーチンは、エラーを適切なレベルで伝播させて処理できる強力なデータ処理パイプラインを形成することができます。これにより、複雑なデータ処理システムを構築しやすくなります。
4. ジェネレータの finally ブロックは、ジェネレータがどのように終了してもクリーンアップ操作が実行されることを保証します。これにより、プログラムの整合性を維持し、リソースリークを防ぐことができます。

まとめ

この実験では、Python のジェネレータとコルーチンにおける yield 文を管理するための重要なテクニックを学びました。ジェネレータのライフタイム管理について調査し、クローズ時やガベージコレクション時の GeneratorExit 例外の処理、および反復処理の中断と再開の制御を行いました。さらに、ジェネレータでの例外処理について学び、throw() メソッドの使用や、例外を適切に処理する堅牢なジェネレータの作成方法を学びました。

これらのテクニックは、堅牢で保守可能な Python アプリケーションを構築するための基礎となります。データ処理、非同期操作、およびリソース管理に役立ちます。ジェネレータのライフタイムを適切に管理し、例外を処理することで、エラーを適切に処理し、不要になったリソースをクリーンアップする弾力性のあるシステムを作成することができます。