Получение результатов внешних команд через subprocess в Python

Расширенные примеры работы с выводом subprocess

Чтение вывода с таймаутом при помощи Popen

Для предотвращения зависания можно использовать метод wait(timeout=...) или communicate(timeout=...). Пример с timeout и обработкой исключения TimeoutExpired:

import subprocess
import time

try:
    process = subprocess.Popen(['sleep', '10'], stdout=subprocess.PIPE, text=True)
    stdout, stderr = process.communicate(timeout=3)
    print('Вывод:', stdout)
except subprocess.TimeoutExpired:
    process.kill()  # принудительное завершение
    print('Процесс превысил время ожидания и был завершен')

Процесс превысил время ожидания и был завершен

Комбинированный вывод stdout и stderr в реальном времени с метками

Для разделения сообщений из разных потоков можно запустить два потока чтения. В примере используются потоки и очередь для асинхронного сбора:

import subprocess
import threading
import queue

def reader(stream, q, name):
    for line in iter(stream.readline, ''):
        q.put(f'{name}: {line.strip()}')
    stream.close()

process = subprocess.Popen(
    ['bash', '-c', 'echo stdout1; echo stderr1 >&2; echo stdout2'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

q = queue.Queue()
th1 = threading.Thread(target=reader, args=(process.stdout, q, 'STDOUT'))
th2 = threading.Thread(target=reader, args=(process.stderr, q, 'STDERR'))
th1.start()
th2.start()
th1.join()
th2.join()
process.wait()

while not q.empty():
    print(q.get())

STDOUT: stdout1
STDERR: stderr1
STDOUT: stdout2

Перехват вывода с помощью временного файла

Если объем вывода очень велик, лучше направить его в файл, а не в память. subprocess позволяет использовать файловые объекты:

import subprocess
import tempfile

with tempfile.NamedTemporaryFile(mode='w+', delete=False) as f:
    process = subprocess.Popen(['find', '/usr', '-name', '*.py'], stdout=f, stderr=subprocess.STDOUT)
    process.wait()
    f.seek(0)
    # Можно читать построчно или весь вывод
    count = sum(1 for _ in f)
    print(f'Найдено {count} файлов .py')

Найдено 245 файлов .py

Использование subprocess.DEVNULL для игнорирования вывода

Если вывод не нужен, его можно направить в subprocess.DEVNULL для экономии ресурсов. Это часто используется в командах, где важен только код возврата:

import subprocess
result = subprocess.run(['rm', '-rf', '/tmp/test_dir'], capture_output=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
if result.returncode == 0:
    print('Удаление прошло успешно')

Удаление прошло успешно

Продвинутая работа с буферизацией: отключение буферизации через PYTHONUNBUFFERED

При запуске интерпретатора Python через subprocess часто возникает проблема с буферизацией вывода. Установка переменной окружения PYTHONUNBUFFERED=1 отключает буферизацию:

import subprocess
import os

env = os.environ.copy()
env['PYTHONUNBUFFERED'] = '1'
process = subprocess.Popen(
    ['python3', '-c', 'import sys; print("строка 1"); sys.stderr.write("ошибка\n")'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True,
    env=env
)
stdout, stderr = process.communicate()
print('stdout:', stdout)
print('stderr:', stderr)

stdout: строка 1
stderr: ошибка

Получение вывода с помощью асинхронного subprocess (asyncio.create_subprocess_exec)

В асинхронном коде удобно использовать asyncio.create_subprocess_exec. Вывод считывается через корутины:

import asyncio

async def get_output():
    process = await asyncio.create_subprocess_exec(
        'git', 'log', '--oneline', '-1',
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    stdout, stderr = await process.communicate()
    print('Последний коммит:', stdout.decode().strip())

asyncio.run(get_output())

Последний коммит: a1b2c3d Исправление бага

Парсинг вывода команды df с помощью subprocess

Пример сложного анализа вывода: получение информации о дисковом пространстве и преобразование в словарь:

import subprocess
import shlex

result = subprocess.run(['df', '-h', '--output=target,size,used,avail,pcent'], capture_output=True, text=True)
lines = result.stdout.strip().split('\n')
headers = lines[0].split()
data = []
for line in lines[1:]:
    parts = line.split(maxsplit=4)
    if len(parts) == 5:
        data.append(dict(zip(headers, parts)))
print(data)

[{'Mounted': '/', 'Size': '27G', 'Used': '15G', 'Avail': '11G', 'Use%': '59%'}, {'Mounted': '/home', 'Size': '50G', 'Used': '20G', 'Avail': '30G', 'Use%': '40%'}]