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 ()