pyterminateダウンロード - pyterminateソースコードのダウンロード

pyterminate

AI ソースコード

v0.0.8

ダウンロード

pyterminate

プログラム終了時にクリーンアップコードを確実に実行します。

なぜこれが存在するのでしょうか?

現在、Python で終了動作を処理するための 2 つの組み込みモジュール ( atexitとsignalがあります。ただし、これらを直接使用すると、多くの定型コードが繰り返され、誤って間違いやすいいくつかの明白ではない動作が発生するため、このパッケージを作成しました。

atexitモジュールは信号を処理できないため、現時点では不十分です。通常の終了または例外による終了を処理できないため、 signalモジュールは現在不十分です。

典型的なアプローチには、 atexitと目的の信号の両方で関数を登録する頻繁に繰り返されるコードが含まれます。ただし、関数が 2 回実行されないように (または冪等であり)、以前に登録されたシグナルハンドラーが呼び出されるようにするために、特別な注意が必要な場合があります。

何ができるのでしょうか？

このパッケージは、次の動作を実行または許可します。

プログラム終了時に呼び出される関数を登録する
- 常に正常終了または例外による終了: @ pyterminate .register
- 任意の信号に合わせて構成可能:
  @ pyterminate .register(signals=(signal.SIGINT, signal.SIGABRT))
複数の機能を登録可能
以前に登録されたシグナルハンドラーを呼び出します
キャプチャされた信号でゼロまたはゼロ以外の終了コードを許可します。
@ pyterminate .register(successful_exit=True)
SIGINTでのKeyboardInterruptの抑制またはスローを許可します。
@ pyterminate .register(keyboard_interrupt_on_sigint=True)
- 追加の例外処理が定義されている場合は、 KeyboardInterruptスローするとよいでしょう。
関数の登録を解除できるようにします: pyterminate .unregister(func)
登録された関数の実行中は要求されたシグナルを無視し、中断されないようにします。
- SIGKILLとos._exit()の呼び出しは無視できないことに注意することが重要です。

クイックスタート

python3 -m pip install pyterminate

pyterminate @ pyterminate .register( args=(None,), kwargs={"b": 42}, signals=(signal.SIGINT, signal.SIGTERM), successful_exit=True, keyboard_interrupt_on_sigint=True ) def cleanup(*args, **kwargs): ... # or pyterminate .register(cleanup, ...)">

 import signal

import pyterminate


@ pyterminate . register (
    args = ( None ,),
    kwargs = { "b" : 42 },
    signals = ( signal . SIGINT , signal . SIGTERM ),
    successful_exit = True ,
    keyboard_interrupt_on_sigint = True
)
def cleanup ( * args , ** kwargs ):
    ...

# or

pyterminate . register ( cleanup , ...)

ヒント、コツ、その他の注意事項

フォーク後の重複登録

フォークによって新しいプロセスを作成するとプロセス全体が複製されるため、以前に登録された関数もフォークされたプロセスに登録されます。これはフォークの明らかな結果ですが、登録された関数が共有リソースにアクセスしているかどうかを考慮することが重要です。この動作を回避するには、フォークされたプロセスの開始時に関数の登録を解除するか、プロセスの ID に基づいてゲートするか、その他の適切な同期方法を使用します。

マルチプロセスの起動方法

Python のmultiprocessingモジュールでプロセスを開始する場合、 forkメソッドは終了時に登録された関数の呼び出しに失敗します。これは、プロセスが内部でos._exit()によって終了され、すべてのクリーンアップがバイパスされ、プロセスが即時に強制終了されるためです。

これを回避する 1 つの方法は、アプリケーションで許容できる場合は、 "spawn"開始メソッドを使用することです。もう 1 つの方法は、関数をユーザー定義シグナルに登録し、プロセスコードを try-excel ブロックでラップし、最後にユーザー定義シグナルを発生させることです。 pyterminateこの機能をexit_with_signalデコレータの形式で提供します。このデコレータは、装飾された関数を try-finally ブロックで単純にラップし、指定されたシグナルを発生させます。使用例:

pyterminate @ pyterminate .exit_with_signal(signal.SIGUSR1) def run_process(): @ pyterminate .register(signals=[signal.SIGUSR1, signal.SIGINT, signal.SIGTERM]) def cleanup(): ... ... if __name__ == "__main__" mp.set_start_method("fork") proc = mp.Process(target=run_process) proc.start() try: proc.join(timeout=300) except TimeoutError: proc.terminate() proc.join()">

 import multiprocessing as mp
import signal

import pyterminate


@ pyterminate . exit_with_signal ( signal . SIGUSR1 )
def run_process ():

    @ pyterminate . register ( signals = [ signal . SIGUSR1 , signal . SIGINT , signal . SIGTERM ])
    def cleanup ():
        ...

    ...


if __name__ == "__main__"
    mp . set_start_method ( "fork" )

    proc = mp . Process ( target = run_process )
    proc . start ()

    try :
        proc . join ( timeout = 300 )
    except TimeoutError :
        proc . terminate ()
        proc . join ()