Subprocess.Popen: примеры (PYTHON)

Примеры использования

Пример выполнения простой команды с захватом вывода:

import subprocess
process = subprocess.Popen(['ls', '-l'], stdout=subprocess.PIPE, text=True)
output, errors = process.communicate()
print('Код возврата:', process.returncode)
print('Вывод:', output)

Код возврата: 0
Вывод: total 4
-rw-r--r-- 1 user group 123 Apr 10 10:00 file.txt

Пример с перенаправлением вывода в файл:

with open('output.txt', 'w') as f:
    subprocess.Popen(['echo', 'Hello World'], stdout=f)

Пример использования shell-пайпов:

process = subprocess.Popen('ls -la | grep .py', shell=True, stdout=subprocess.PIPE, text=True)
print(process.communicate()[0])

Альтернативные функции в Python

subprocess.run(): Рекомендуемая функция для большинства случаев. Упрощенный интерфейс, автоматически ожидает завершения процесса и возвращает объект CompletedProcess.

result = subprocess.run(['ls', '-l'], capture_output=True, text=True)
print(result.stdout)

subprocess.call(): Выполняет команду и возвращает код возврата.

subprocess.check_output(): Выполняет команду, захватывает вывод и возвращает его. В случае ошибки вызывает исключение.

os.system(): Простой способ выполнить команду, но с ограниченными возможностями управления процессом.

Аналоги в других языках программирования

PHP: proc_open() и shell_exec().

$descriptorspec = [0 => ['pipe', 'r'], 1 => ['pipe', 'w']];
$process = proc_open('ls -l', $descriptorspec, $pipes);
echo stream_get_contents($pipes[1]);
proc_close($process);

JavaScript (Node.js): child_process.spawn().

const { spawn } = require('child_process');
const ls = spawn('ls', ['-l']);
ls.stdout.on('data', (data) => {
  console.log(data.toString());
});

Java: ProcessBuilder и Runtime.exec().

ProcessBuilder pb = new ProcessBuilder('ls', '-l');
Process process = pb.start();
BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()));
String line;
while ((line = reader.readLine()) != null) {
    System.out.println(line);
}

C#: System.Diagnostics.Process.

using System.Diagnostics;
var process = new Process();
process.StartInfo.FileName = 'ls';
process.StartInfo.Arguments = '-l';
process.StartInfo.UseShellExecute = false;
process.StartInfo.RedirectStandardOutput = true;
process.Start();
Console.WriteLine(process.StandardOutput.ReadToEnd());

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

Использование shell=True без экранирования пользовательского ввода может привести к уязвимостям инъекции команд.

user_input = 'file.txt; rm -rf /'
subprocess.Popen(f'cat {user_input}', shell=True) # Опасный код!

Блокировка из-за заполненных буферов при неправильном порядке чтения потоков.

process = subprocess.Popen(['program'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
# Неправильно: сначала читаем stderr, а stdout может быть заполнен
stderr = process.stderr.read()
stdout = process.stdout.read()

Игнорирование кодов возврата может привести к незамеченным ошибкам.

Изменения в последних версиях

В Python 3.7 добавлен аргумент text как синоним для universal_newlines.

В Python 3.8 появился аргумент errors для обработки ошибок кодировки в текстовом режиме.

В Python 3.9 добавлен аргумент capture_output для упрощения захвата стандартных потоков.

В Python 3.12 улучшено поведение подпроцессов в Windows при использовании консоли.

Расширенные примеры

Цепочка процессов с использованием пайпов:

import subprocess
p1 = subprocess.Popen(['ls', '-la'], stdout=subprocess.PIPE)
p2 = subprocess.Popen(['grep', '.py'], stdin=p1.stdout, stdout=subprocess.PIPE, text=True)
p1.stdout.close()
output = p2.communicate()[0]
print(output)

Одновременное чтение stdout и stderr с таймаутом:

import subprocess
from threading import Thread
import time

def read_stream(stream, result):
    result.append(stream.read())

process = subprocess.Popen(['long_running_script.sh'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
out, err = [], []
t1 = Thread(target=read_stream, args=(process.stdout, out))
t2 = Thread(target=read_stream, args=(process.stderr, err))
t1.start()
t2.start()

# Ждем завершения с таймаутом
for i in range(10):
    if process.poll() is not None:
        break
    time.sleep(1)
else:
    process.terminate()
    process.wait()

t1.join()
t2.join()
print('stdout:', ''.join(out))
print('stderr:', ''.join(err))

Использование переменных окружения:

import subprocess, os
new_env = os.environ.copy()
new_env['MY_VAR'] = 'custom_value'
process = subprocess.Popen(['env'], env=new_env, stdout=subprocess.PIPE, text=True)
output = process.communicate()[0]
print('MY_VAR' in output)  # True