es.davy.ai

Preguntas y respuestas de programación confiables

¿Tienes una pregunta?

Si tienes alguna pregunta, puedes hacerla a continuación o ingresar lo que estás buscando.

Cómo terminar un proceso de Python y entrar en el bloque ‘finally’ en una declaración ‘try-except’ de Python.

Tengo dos archivos de Python: script.py y counting.py.

counting.py

from time import sleep

try:
    for n in range(1, 10):
        print('contando', n)
        sleep(1)
except Exception as e:
    print('el programa terminó con una excepción', e)
else:
    print('el programa ha finalizado')
finally:
    sleep(4)
    print('este es el bloque finally')

script.py

import sys
import pathlib
import subprocess
import signal

en_windows = True if sys.platform == 'win32' else False
en_linux = True if sys.platform == 'linux' else False

if en_windows:
    process = subprocess.Popen(
        ['python', 'counting.py'],
        cwd=pathlib.Path(__file__).parent.resolve(),
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        universal_newlines=True,
        creationflags=subprocess.CREATE_NEW_PROCESS_GROUP,
    )
elif en_linux:
    process = subprocess.Popen(
        ['python', 'counting.py'],
        cwd=pathlib.Path(__file__).parent.resolve(),
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        universal_newlines=True,
        preexec_fn=os.setsid,
    )
try:
    out, err = process.communicate(timeout=4)
    print(out, err)
except subprocess.TimeoutExpired:
    if en_windows:
        process.send_signal(signal.CTRL_BREAK_EVENT)
        process.kill()
    elif en_linux:
        os.killpg(process.pid, signal.SIGTERM)
    print('tiempo agotado')

Cuando ejecuto <code>script.py</code>, la salida es:

tiempo agotado

El proceso devolvió 0 (0x0) tiempo de ejecución: 4.778 s


<strong>¿Cómo puedo terminar el <code>proceso</code> y entrar en el bloque <code>finally</code> en el archivo <code>counting.py</code>?</strong> P.D. Cuando ejecuto <code>counting.py</code> en la terminal, el programa entra en el bloque <code>finally</code> después de presionar <code>CTRL+C</code> en el teclado:

python counting.py
contando 1
contando 2
contando 3
este es el bloque finally
Traceback (most recent call last):
File “counting.py”, line 6, in
sleep(1)
KeyboardInterrupt
^C

“`
¿Cómo puedo lograr esto utilizando el módulo subprocess y matando el proceso hijo?
He probado los códigos anteriores en Windows (Windows 10) y Linux (CentOS 8).

Tags:  , , ,

Answer

  1. Avatar for davy.ai

    Para terminar el proceso creado en script.py y entrar en el bloque finally en counting.py, podemos modificar el bloque except subprocess.TimeoutExpired de la siguiente manera:

    except subprocess.TimeoutExpired:
        if on_windows:
            # enviar un evento CTRL+C al proceso creado
            process.send_signal(signal.CTRL_C_EVENT)
            process.wait()
        elif on_linux:
            # enviar una señal SIGINT al grupo de procesos
            os.killpg(os.getpgid(process.pid), signal.SIGINT)
            process.wait()
        print('tiempo de espera agotado')
    

    Esto enviará una señal adecuada al proceso creado, causando que termine con una excepción KeyboardInterrupt, la cual será capturada por el bloque except en counting.py, y el programa entrará en el bloque finally como se desea.

    Es importante destacar que estamos llamando a process.wait() después de enviar la señal, lo cual hará que el proceso padre espere hasta que el proceso hijo haya terminado antes de continuar con la ejecución.

    Además, para enviar un evento CTRL+C al proceso creado en Windows, debemos usar la señal CTRL_C_EVENT, en lugar de CTRL_BREAK_EVENT que se estaba utilizando en el código original.

    Con estas modificaciones, el archivo completo script.py se verá así:

    import sys
    import pathlib
    import subprocess
    import signal
    import os
    
    on_windows = True if sys.platform == 'win32' else False
    on_linux = True if sys.platform == 'linux' else False
    
    if on_windows:
        process = subprocess.Popen(
            ['python', 'counting.py'],
            cwd=pathlib.Path(__file__).parent.resolve(),
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            universal_newlines=True,
            creationflags=subprocess.CREATE_NEW_PROCESS_GROUP,
        )
    elif on_linux:
        process = subprocess.Popen(
            ['python', 'counting.py'],
            cwd=pathlib.Path(__file__).parent.resolve(),
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            universal_newlines=True,
            preexec_fn=os.setsid,
        )
    try:
        out, err = process.communicate(timeout=4)
        print(out, err)
    except subprocess.TimeoutExpired:
        if on_windows:
            # enviar un evento CTRL+C al proceso creado
            process.send_signal(signal.CTRL_C_EVENT)
            process.wait()
        elif on_linux:
            # enviar una señal SIGINT al grupo de procesos
            os.killpg(os.getpgid(process.pid), signal.SIGINT)
            process.wait()
        print('tiempo de espera agotado')
    

    Una vez que el proceso sea terminado enviando la señal apropiada, el bloque finally en counting.py será ejecutado como se espera.

Comments are closed.