Python как инструмент для управления системными процессами от subprocess до psutil

Раздел: Системное администрирование -> Запуск и завершение процессов

Управление выполнением программ в Python для системного администрирования

Как запустить внешнюю программу и получить её стандартный вывод с контролем ошибок?

Основным и наиболее рекомендуемым способом является использование модуля subprocess, в частности функции subprocess.run. Она позволяет запустить команду, дождаться её завершения и получить результаты.

import subprocess
result = subprocess.run(['ls', '-l', '/tmp'], capture_output=True, text=True, timeout=10)
print(result.stdout)

Python call program (управление выполнением python программ)

Параметр capture_output объединяет stdout и stderr в атрибуты result.stdout и result.stderr. Параметр text=True возвращает строки вместо байтов. timeout ограничивает время выполнения, при превышении выбрасывается исключение TimeoutExpired.

Типичные ошибки:

Забыв указать text=True, получают байтовый вывод, который нужно декодировать вручную.
При использовании shell=True с аргументом-строкой можно получить проблемы с пробелами и инъекцией. Без необходимости следует передавать список аргументов.
Если процесс генерирует большой объём данных, чтение через capture_output может привести к нехватке памяти. В таких случаях лучше использовать Popen с потоковой обработкой.

Цель и случаи использования: запуск быстрых команд, сбор их вывода для последующего анализа, автоматизация действий системного администратора (например, пинг, проверка занятости диска).

Как выполнить команду в оболочке без необходимости обрабатывать вывод?

Модуль os содержит устаревшую функцию system, которая выполняет команду через оболочку и возвращает код завершения. Вывод команды направляется непосредственно в stdout/err терминала Python.

import os
ret = os.system("echo 'Hello'")
print('Return code:', ret)

Проблемы: невозможно получить вывод программно, отсутствует контроль над таймаутом, уязвимость для shell injection при передаче пользовательских данных. Этот вариант подходит только для простых команд, когда вывод не нужен, или для быстрых скриптов без требований безопасности.

На платформе Windows os.system может вести себя иначе, чем на Unix. Для кросс-платформенных решений рекомендуется subprocess.

Как организовать потоковое чтение вывода процесса в реальном времени?

Использование subprocess.Popen с перенаправлением stdout позволяет построчно обрабатывать вывод по мере его генерации.

import subprocess
proc = subprocess.Popen(['ping', '-c', '4', 'google.com'], stdout=subprocess.PIPE, text=True)
for line in proc.stdout:
    print('Получена строка:', line.strip())
proc.wait()

Важно: при таком подходе необходимо читать все данные из stdout, иначе процесс может зависнуть при заполнении буфера. Также следует обрабатывать stderr отдельно, если требуется.

Если процесс не закрывает stdout (например, интерактивная программа), цикл будет бесконечным. В таком случае нужно использовать таймаут или читать определённое количество строк.

Как запустить процесс в фоновом режиме и позднее проверить его статус?

Экземпляр Popen без вызова wait позволяет продолжить работу, а затем опросить состояние через poll или принудительно завершить через terminate/kill.

import subprocess, time
proc = subprocess.Popen(['sleep', '10'])
print('Процесс запущен, PID:', proc.pid)
time.sleep(3)
if proc.poll() is None:
    print('Процесс ещё работает, завершаем')
    proc.terminate()
    proc.wait()
else:
    print('Процесс завершился')

Такой подход применяется для запуска долгих задач, мониторинга их выполнения и принудительного завершения по таймауту или условию.

При завершении родительского процесса дочерние процессы могут остаться висеть (зомби). Для предотвращения следует всегда вызывать wait или использовать контекстный менеджер Popen.

Как запустить несколько программ параллельно и дождаться их завершения?

Модуль concurrent.futures в сочетании с subprocess упрощает параллельное выполнение команд.

import subprocess
from concurrent.futures import ThreadPoolExecutor

def run_cmd(cmd):
    return subprocess.run(cmd, capture_output=True, text=True)

cmds = [['echo', 'hello'], ['uname', '-a'], ['date']]
with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(run_cmd, cmds))
for res in results:
    print('stdout:', res.stdout.strip())

Цель: ускорение выполнения набора независимых команд, например, сбор информации с нескольких серверов через SSH.

Использование потоков для CPU-ёмких задач не даст прироста из-за GIL. Для таких случаев следует применять ProcessPoolExecutor. Однако запуск внешних процессов через subprocess уже выполняется в отдельных процессах, поэтому ThreadPoolExecutor часто достаточен.

Расширенные примеры управления процессами

Пример 1: Полный контроль над запуском с обработкой ошибок

Пример

import subprocess
try:
    result = subprocess.run(
        ['grep', 'error', '/var/log/syslog'],
        capture_output=True,
        text=True,
        check=True,
        timeout=30
    )
    print('Найдены строки:')
    print(result.stdout)
except subprocess.CalledProcessError as e:
    print('Команда завершилась с кодом', e.returncode)
    print('stderr:', e.stderr)
except subprocess.TimeoutExpired:
    print('Время истекло')

(пример гипотетического вывода)
Найдены строки:
Jan 25 10:15:22 host kernel: [12345] error in module

Пример 2: Потоковая обработка большого вывода с записью в файл

Пример

import subprocess
with open('output.txt', 'w') as f:
    proc = subprocess.Popen(['dmesg'], stdout=subprocess.PIPE, text=True)
    for line in proc.stdout:
        f.write(line)
        if 'usb' in line:
            print('Найдена USB-строка:', line.strip())
    proc.wait()

(строки из dmesg, записанные в файл, и вывод на консоль)
Найдена USB-строка: usb 1-1: new high-speed USB device number 2 using ehci_hcd

Пример 3: Параллельный запуск с помощью ProcessPoolExecutor

Пример

import subprocess
from concurrent.futures import ProcessPoolExecutor

def check_host(host):
    result = subprocess.run(['ping', '-c', '1', host], capture_output=True, text=True, timeout=5)
    if result.returncode == 0:
        return f'{host} доступен'
    else:
        return f'{host} недоступен'

hosts = ['google.com', 'github.com', 'example.com']
with ProcessPoolExecutor() as executor:
    for host, status in zip(hosts, executor.map(check_host, hosts)):
        print(status)

google.com доступен
github.com доступен
example.com недоступен

Пример 4: Отправка сигналов процессу и его потомкам

Пример

import subprocess, os, signal, time
script = 'import time, subprocess\nproc = subprocess.Popen([\'sleep\', \'20\'])\ntime.sleep(10)\nproc.wait()'
proc = subprocess.Popen(['python3', '-c', script], start_new_session=True)
time.sleep(2)
os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
proc.wait()
print('Родительский процесс завершён')

(через 2 секунды процесс завершается)
Родительский процесс завершён

Пример 5: Использование psutil для поиска и завершения процесса по имени

Пример

import psutil

for proc in psutil.process_iter(['pid', 'name']):
    if proc.info['name'] == 'sleep':
        print('Найден процесс', proc.info['pid'])
        proc.terminate()
        proc.wait(timeout=5)
        print('Завершён')

Найден процесс 12345
Завершён

Пример 6: Запуск с пользовательским окружением и рабочим каталогом

Пример

import subprocess, os
my_env = os.environ.copy()
my_env['MY_VAR'] = 'value123'
result = subprocess.run(
    ['printenv', 'MY_VAR'],
    capture_output=True,
    text=True,
    env=my_env,
    cwd='/tmp'
)
print('Значение MY_VAR:', result.stdout.strip())

Значение MY_VAR: value123

Управление выполнением Python программ - comments

Python call program (python)