Obsługa subprocesów w Pythonie

Podziel się tym postem!

Obsługa subprocesów w Pythonie do najprzyjaźniejszych nie należy.

Dokumentacja zawiera dużo informacji (https://docs.python.org/3/library/subprocess.html) natomiast brakuje pełnego przykładu, wraz z bezpieczną obsługą wyjątków.

Najprostsze zastosowanie pokazuje poniższy kod:

import subprocess
proc = subprocess.Popen(
   ["npm", "install"], cwd=tmp_dir, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
(output, err) = proc.communicate(timeout=120)

Proste, prawda?

Pobierz bezpłatny e-book!

Dowiedz się więcej o kluczowych kompetencjach i narzędziach dla początkującego programisty Pythona i zyskaj cenne informacje.

Pobierz teraz

Problem się pojawia z przekroczeniem czasu - Timeout. W tym przypadku logi wykonanych operacji nie są zebrane a sam proces "wisi".

Pewnym udoskonaleniem jest łapanie wyjątku:

import subprocess
proc = subprocess.Popen(...)
try:
    outs, errs = proc.communicate(timeout=15)
except TimeoutExpired:
    proc.kill()
    outs, errs = proc.communicate()

W tym przypadku już mamy wszystkie dane wyjściowe oraz posprzątaliśmy popsuty proces. To już pozwoli na obsługę TimeoutExpired w dalszym kodzie, natomiast nie złapie pozostałych wyjątków, ani ich szczegółów.

Najlepszym rozwiązaniem wydaje się zbudowanie funkcji do obsługi.

import subprocess
import sys

def runInSubprocess(*args, cwd=None, timeout=None, check=True):
    with subprocess.Popen(args, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as process:
        try:
            stdout, stderr = process.communicate(input, timeout=timeout)
        except subprocess.TimeoutExpired:
            process.kill() # cleanup
            stdout, stderr = process.communicate() # grab error message and progress
            raise subprocess.TimeoutExpired(
                process.args, timeout, output=stdout, stderr=stderr,
            ) # raise error again, will all details for parent to handle.
        except Exception:
            process.kill()
            process.wait() # wait for process to die
            raise # let parent handle this
        retcode = process.poll() # verify if process terminated
        if check and retcode:
            raise subprocess.CalledProcessError(
                retcode, process.args, output=stdout, stderr=stderr,
            ) # let parent handle this
                # return process information
        return subprocess.CompletedProcess(process.args, retcode, stdout, stderr)

Powyższy kod pozwala na pełną obsługę subprocess, w przypadku TimeoutExpired zwróci zarówno ewentualny błąd jak i postęp zadania. W przypadku pozostałych wyjątków "pozbiera" dostępne informacje i zwróci w postaci wyjątku. To z kolei pozwala na łatwą i przejrzystą obsługę w procesie wywołującym, w postaci, na przykład retry.

import subprocess

def subprocess_retry(*args, cwd=None, timeout=None, check=True)
    retry_attempts = 3
    exception_to_reraise = None
    for attempt in range(retry_attempts):
        try:
            runInSubprocess(*args, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        except Exception as e:
            exception_to_reraise = e
            print(f"<<< FAILED attempt {attempt}")
        else:
            break
    else:
        print(f"<<< FAILED All ({retry_attempts}) attempts")
        raise exception_to_reraise

Masz uwagi do posta, chcesz porozmawiać, szukasz pomocy z Pythonem i Django? Napisz do mnie!

Ps. Spodobał Ci się post? Udostępnij go na swoich kanałach.