pyterminate

pyterminate

Ejecute de manera confiable el código de limpieza al finalizar el programa.

Tabla de contenido

• ¿Por qué existe esto?
• ¿Qué puede hacer?
• Inicio rápido
• Consejos, trucos y otras notas.
  • Registro duplicado después de la bifurcación
  • Método de inicio de multiprocesamiento

¿Por qué existe esto?

Actualmente hay dos módulos integrados para manejar el comportamiento de terminación en Python: atexit y signal . Sin embargo, usarlos directamente genera una gran cantidad de código repetitivo repetido y algunos comportamientos no obvios que pueden ser fáciles de equivocar accidentalmente, razón por la cual escribí este paquete.

El módulo atexit actualmente es insuficiente ya que no maneja señales. El módulo signal actualmente es insuficiente ya que no maneja salidas normales o provocadas por excepciones.

Los enfoques típicos incluirían un código repetido con frecuencia que registre una función tanto con atexit como con las señales deseadas. Sin embargo, a veces es necesario tener especial cuidado para garantizar que la función no se ejecute dos veces (o sea idempotente) y que se llame a un controlador de señales previamente registrado.

¿Qué puede hacer?

Este paquete hace o permite el siguiente comportamiento:

• Registrar una función para ser llamada al finalizar el programa

  • Siempre en terminación normal o causada por una excepción: @ pyterminate .register
  • Configurable para cualquier señal deseada:
    @ pyterminate .register(signals=(signal.SIGINT, signal.SIGABRT))

• Permite registrar múltiples funciones

• Llamará a los manejadores de señales registrados anteriormente.

• Permite códigos de salida cero o distintos de cero en señales capturadas:
  @ pyterminate .register(successful_exit=True)

• Permite suprimir o lanzar KeyboardInterrupt en SIGINT :
  @ pyterminate .register(keyboard_interrupt_on_sigint=True)

  • Es posible que desees lanzar un KeyboardInterrupt si hay definido un manejo de excepciones adicional.

• Permite cancelar el registro de funciones: pyterminate .unregister(func)

• Ignore las señales solicitadas mientras se ejecuta la función registrada, asegurándose de que no se interrumpa.

  • Es importante tener en cuenta que SIGKILL y las llamadas a os._exit() no se pueden ignorar.

Inicio rápido

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 , ...)

Consejos, trucos y otras notas.

Registro duplicado después de bifurcar

Dado que la creación de un nuevo proceso mediante la bifurcación duplica todo el proceso, cualquier función registrada previamente también se registrará en el proceso bifurcado. Esta es una consecuencia obvia de la bifurcación, pero es importante considerarla si las funciones registradas acceden a recursos compartidos. Para evitar este comportamiento, puede cancelar el registro de la función al comienzo del proceso bifurcado, realizar una puerta basada en el ID del proceso o utilizar cualquier otro método de sincronización que sea apropiado.

Método de inicio de multiprocesamiento

Al iniciar procesos con el módulo de multiprocessing de Python, el método fork no podrá llamar a las funciones registradas al salir, ya que el proceso finaliza con os._exit() internamente, lo que omite toda la limpieza y finaliza inmediatamente el proceso.

Una forma de solucionar esto es utilizar el método de inicio "spawn" si es aceptable para su aplicación. Otro método es registrar su función en una señal definida por el usuario y envolver su código de proceso en el bloque try-except, generando la señal definida por el usuario al final. pyterminate proporciona esta funcionalidad en forma del decorador exit_with_signal , que simplemente envuelve la función decorada en un bloque try-finally y genera la señal dada. Uso de ejemplo:

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