Когда вызов внешней программы в Python завершается ошибкой
Ошибки подпроцесса в Python: причины и решения
Как гарантированно обработать все типичные ошибки при запуске внешней команды?
Наиболее эффективный способ - использовать функцию subprocess.run() с параметром check=True и предварительной проверкой доступности исполняемого файла через shutil.which(). Такой подход перехватывает исключения CalledProcessError (ненулевой код возврата), FileNotFoundError (программа не найдена) и TimeoutExpired (превышение времени ожидания).
import subprocess
import shutil
import sys
def run_command(cmd, timeout=None):
executable = cmd[0] if isinstance(cmd, list) else cmd.split()[0]
# Проверка существования исполняемого файла
if shutil.which(executable) is None:
raise FileNotFoundError(f"Исполняемый файл '{executable}' не найден в PATH")
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=True,
timeout=timeout
)
return result.stdout
except subprocess.CalledProcessError as e:
sys.stderr.write(f"Команда завершилась с кодом {e.returncode}\n")
sys.stderr.write(f"stdout: {e.stdout}\n")
sys.stderr.write(f"stderr: {e.stderr}\n")
raise
except subprocess.TimeoutExpired:
sys.stderr.write(f"Команда превысила время ожидания {timeout} сек\n")
raise
except FileNotFoundError:
sys.stderr.write(f"Исполняемый файл не найден\n")
raise
Python subprocess error (ошибка подпроцесса в python)
Пояснение: аргумент capture_output=True перехватывает stdout и stderr, text=True возвращает строки, а не байты. timeout предотвращает зависание. В случае ошибки вызывается исключение, которое затем обрабатывается в вызывающем коде.
Типичная ошибка: забыть указать check=True - тогда при ненулевом коде возврата исключение не возникнет, и программа продолжит работу с некорректными данными. Решение: всегда проверять result.returncode вручную или использовать check=True.
Другая ошибка: передача команды в виде строки вместо списка при shell=False (по умолчанию) может привести к FileNotFoundError, так как строка интерпретируется как имя одного файла. Решение: всегда разбивать команду на список аргументов.
Какие альтернативные способы вызова подпроцесса существуют и в каких случаях они применяются?
Вариант 1: Использование shell=True для простых команд
Параметр shell=True позволяет передавать команду как единую строку, которая выполняется оболочкой. Удобно для быстрых однострочников, но повышает риск внедрения команд (injection).
import subprocess
# Небезопасный пример (shell=True)
result = subprocess.run('echo $HOME', shell=True, capture_output=True, text=True)
print(result.stdout.strip())
Проблема: если пользователь может контролировать часть ввода, злоумышленник может выполнить произвольную команду. Решение: избегать shell=True с непроверенными данными. Использовать только для фиксированных, заранее известных команд.
Вариант 2: Асинхронный запуск через subprocess.Popen
Когда требуется взаимодействовать с процессом в реальном времени (например, читать вывод построчно), используют Popen.
import subprocess
proc = subprocess.Popen(
['ping', '-c', '4', 'example.com'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
try:
stdout, stderr = proc.communicate(timeout=10)
print('stdout:', stdout)
print('stderr:', stderr)
except subprocess.TimeoutExpired:
proc.kill()
stdout, stderr = proc.communicate()
Ошибка: если не вызвать communicate(), процесс может зависнуть. Решение: всегда вызывать communicate() или организовать чтение потоков. Не забыть про timeout для защиты от зависания.
Вариант 3: Использование subprocess.check_output для получения вывода
Функция check_output возвращает stdout при успехе, иначе бросает CalledProcessError. Подходит для простых случаев, когда stderr не нужен.
import subprocess
try:
output = subprocess.check_output(['cat', '/etc/hostname'], text=True)
print('Имя хоста:', output.strip())
except subprocess.CalledProcessError as e:
print('Ошибка:', e)
Проблема: нельзя получить stderr, если команда завершилась с ошибкой. Решение: использовать subprocess.run с capture_output=True для полного контроля.
Вариант 4: Обработка ошибок кодировки вывода
Если внешняя программа выдает байты, несовместимые с кодировкой text=True, возникает UnicodeDecodeError. Решение: отказаться от text=True и обработать байты самостоятельно.
import subprocess
result = subprocess.run(['some_program'], capture_output=True)
# Ручное декодирование с заменой ошибок
print(result.stdout.decode('utf-8', errors='replace'))
Типичная ошибка: полагаться на автоматическое декодирование, что ведет к исключению. Решение: использовать параметр errors='replace' при ручном декодировании или указать encoding='utf-8', errors='replace' в subprocess.run.
Вариант 5: Перехват тайм-аута сложной командой
Команды с длительным выполнением (например, резервное копирование) требуют жесткого лимита времени. Параметр timeout в run решает эту задачу.
import subprocess
try:
subprocess.run(['long_script.sh'], timeout=30, check=True)
except subprocess.TimeoutExpired:
print('Команда прервана по тайм-ауту')
Проблема: после TimeoutExpired процесс может остаться висеть в фоне. Решение: в блоке except принудительно убить процесс через result.kill(), если есть объект результата (при использовании Popen). Для run процесс уже завершен, дополнительных действий не требуется.
Ниже приведены расширенные примеры с полным кодом и выводом.
Пример 1: Команда с ненулевым кодом возврата и разбор stderr
import subprocess
try:
result = subprocess.run(
['ls', '--nonexistent'],
capture_output=True,
text=True,
check=True
)
except subprocess.CalledProcessError as e:
print("Команда завершилась с ошибкой")
print("Код возврата:", e.returncode)
print("stdout:", repr(e.stdout))
print("stderr:", repr(e.stderr))
Команда завершилась с ошибкой Код возврата: 2 stdout: '' stderr: 'ls: unrecognized option '--nonexistent'\nTry 'ls --help' for more information.\n'
Пример 2: Использование shell=True с риском инъекции
import subprocess
user_input = "; cat /etc/passwd" # опасная команда
# Ни в коем случае не делать так на практике!
cmd = "echo " + user_input
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
print("Вывод:", result.stdout)
Вывод:
Пример 3: Асинхронное чтение вывода с помощью Popen и потоков
import subprocess
import threading
def read_output(stream, prefix):
for line in iter(stream.readline, ''):
print(f"[{prefix}] {line.strip()}")
proc = subprocess.Popen(
['ping', '-c', '2', '127.0.0.1'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
t1 = threading.Thread(target=read_output, args=(proc.stdout, 'STDOUT'))
t2 = threading.Thread(target=read_output, args=(proc.stderr, 'STDERR'))
t1.start()
t2.start()
t1.join()
t2.join()
proc.wait()
print("Процесс завершен, код:", proc.returncode)
[STDOUT] PING 127.0.0.1 (127.0.0.1) 56(84) bytes of data. [STDOUT] 64 bytes from 127.0.0.1: icmp_seq=1 ttl=64 time=0.039 ms [STDOUT] 64 bytes from 127.0.0.1: icmp_seq=2 ttl=64 time=0.044 ms [STDOUT] [STDOUT] --- 127.0.0.1 ping statistics --- [STDOUT] 2 packets transmitted, 2 received, 0% packet loss, time 1033ms [STDOUT] rtt min/avg/max/mdev = 0.039/0.041/0.044/0.002 ms Процесс завершен, код: 0
Пример 4: Тайм-аут и убийство зависшего процесса
import subprocess
import time
proc = subprocess.Popen(['sleep', '30'], stdout=subprocess.PIPE)
try:
stdout, stderr = proc.communicate(timeout=2)
except subprocess.TimeoutExpired:
print("Процесс не завершился за 2 секунды, убиваем...")
proc.kill()
stdout, stderr = proc.communicate()
print("Процесс убит")
Процесс не завершился за 2 секунды, убиваем... Процесс убит
Пример 5: Проверка наличия исполняемого файла перед запуском
import shutil
import subprocess
cmd = 'nonexistent_program'
if shutil.which(cmd) is None:
print(f"Утилита '{cmd}' не установлена")
else:
result = subprocess.run([cmd], capture_output=True)
Утилита 'nonexistent_program' не установлена