pyterminate下載 - pyterminate原始碼下載

pyterminate

Ai源碼

v0.0.8

下載

pyterminate

在程式終止時可靠地執行清理程式碼。

為什麼會有這個存在？

目前Python中有兩個用於處理終止行為的內建模組： atexit和signal 。然而，直接使用它們會導致大量重複的樣板程式碼，以及一些不明顯的行為，很容易意外出錯，這就是我編寫這個套件的原因。

atexit模組目前還不夠，因為它無法處理訊號。 signal模組目前還不夠，因為它無法處理正常或異常引起的退出。

典型的方法包括頻繁重複的程式碼，透過atexit和所需訊號註冊函數。但是，有時需要格外小心，以確保函數不會運行兩次（或冪等），並且呼叫先前註冊的訊號處理程序。

它能做什麼？

該軟體包執行或允許以下行為：

註冊一個在程式終止時呼叫的函數
- 始終處於正常或異常引起的終止狀態： @ 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()結束，這會繞過所有清理並立即終止進程。

解決這個問題的一種方法是使用"spawn"啟動方法（如果您的應用程式可以接受）。另一種方法是將函數註冊到使用者定義的訊號，並將進程程式碼包裝在 try- except 區塊中，最後引發使用者定義的訊號。 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 ()