Blog

Python subprocess로 시스템 명령어 제어하기

Python 스크립트에서 셸 명령어나 외부 프로그램을 호출해야 하는 상황은 흔하다. os.system()은 출력을 캡처할 수 없고, os.popen()은 에러 처리와 스트림 제어가 번거롭다. 특히 자동화 스크립트에서 외부 명령의 표준출력/표준에러를 각각 받아 로그로 남기거나, 명령 실패 시 예외를 던져 파이프라인을 중단시키고 싶다면 이런 저수준 API로는 코드가 지저분해지기 쉽다. subprocess 모듈은 이 문제를 표준 라이브러리 수준에서 해결한다. 이 글에서는 subprocess.run()을 중심으로 출력 캡처, 에러 처리, 셸 인젝션 방지, 그리고 Popen을 이용한 실시간 스트리밍까지 정리한다.

run()으로 기본 실행

대부분의 경우 subprocess.run() 하나로 충분하다. 명령과 인자를 리스트로 넘기는 것이 기본이다.

import subprocess

result = subprocess.run(["ls", "-la", "/tmp"])
print(result.returncode)  # 0이면 정상 종료

run()은 명령이 끝날 때까지 블로킹되며, 실행 결과를 CompletedProcess 객체로 반환한다. 이 객체는 returncode, stdout, stderr 속성을 갖는다.

출력 캡처와 인코딩

명령의 출력을 파이썬에서 직접 다루려면 capture_output=True와 text=True를 함께 쓴다. text=True가 없으면 stdout/stderr가 bytes로 반환된다.

result = subprocess.run(
    ["git", "log", "-1", "--format=%H"],
    capture_output=True,
    text=True,
)
commit_hash = result.stdout.strip()
print(commit_hash)

시스템 로케일이 UTF-8이 아닌 환경에서 한글이 깨진다면 encoding="utf-8"을 명시적으로 지정한다.

에러 처리: check=True

기본적으로 run()은 명령이 0이 아닌 코드로 종료해도 예외를 던지지 않는다. 실패를 명시적으로 감지하려면 check=True를 주고, 실패 시 CalledProcessError를 예외로 받아 처리한다.

try:
    subprocess.run(["grep", "PATTERN", "nonexistent.txt"], check=True)
except subprocess.CalledProcessError as e:
    print(f"명령 실패: returncode={e.returncode}")
    print(f"stderr: {e.stderr}")

자동화 스크립트에서는 명령이 실패했는데도 모르고 다음 단계로 넘어가는 것이 가장 흔한 실수다. 실패 시 파이프라인을 멈춰야 하는 명령이라면 항상 check=True를 붙이는 편이 안전하다.

shell=True의 위험과 리스트 인자

shell=True를 쓰면 명령을 문자열 그대로 셸에 넘겨 파이프(|)나 리다이렉션(>) 같은 셸 문법을 그대로 쓸 수 있다. 하지만 사용자 입력이 명령 문자열에 섞여 들어가면 셸 인젝션으로 이어질 수 있다.

# 위험한 예: 사용자 입력이 그대로 셸에 전달된다
filename = "report.txt; rm -rf ~"
subprocess.run(f"cat {filename}", shell=True)  # 절대 이렇게 쓰지 말 것

# 안전한 방법: 리스트로 인자를 분리하면 셸을 거치지 않는다
subprocess.run(["cat", filename])

명령을 문자열로만 갖고 있고 리스트로 쪼개기 번거롭다면 shlex.split()을 쓰면 따옴표·공백 처리를 안전하게 대신해준다.

import shlex

cmd = 'rsync -avz --exclude=".git" src/ dest/'
subprocess.run(shlex.split(cmd))

타임아웃과 입력 전달

외부 명령이 멈춰버리는 상황을 대비해 timeout을 지정할 수 있다. 시간 내에 끝나지 않으면 TimeoutExpired 예외가 발생한다. 표준입력에 데이터를 넣어야 한다면 input 인자를 쓴다.

try:
    result = subprocess.run(
        ["some-long-running-tool"],
        timeout=30,
        capture_output=True,
        text=True,
    )
except subprocess.TimeoutExpired:
    print("30초 안에 끝나지 않아 종료했다")

# 표준입력으로 데이터 전달
result = subprocess.run(
    ["sort"],
    input="banana\napple\ncherry\n",
    capture_output=True,
    text=True,
)
print(result.stdout)

Popen으로 실시간 스트리밍 처리

run()은 명령이 끝날 때까지 기다렸다가 결과를 한 번에 돌려준다. 로그를 실시간으로 한 줄씩 처리하고 싶다면 저수준 API인 Popen을 직접 쓴다.

with subprocess.Popen(
    ["journalctl", "-f"],
    stdout=subprocess.PIPE,
    text=True,
) as proc:
    for line in proc.stdout:
        if "ERROR" in line:
            print("감지:", line.strip())

with 구문을 쓰면 블록을 벗어날 때 프로세스 핸들과 파이프가 자동으로 정리된다.

주의사항

stdout과 stderr를 둘 다 PIPE로 캡처하면서 출력량이 많은 명령을 Popen으로 직접 다룰 때는 데드락에 주의해야 한다. 한쪽 파이프 버퍼가 가득 차면 프로세스가 멈추고, 그 상태에서 부모 프로세스가 다른 파이프를 읽으려고 기다리면 서로 블로킹된다. 이런 상황을 피하려면 run()이나 communicate()처럼 두 스트림을 동시에 읽어주는 API를 쓰는 게 안전하다.

os.system()은 내부적으로 셸을 거치기 때문에 shell=True와 같은 위험을 그대로 안고 있고, 출력 캡처도 되지 않는다. 새 코드에서는 os.system()/os.popen() 대신 subprocess를 쓰는 것이 권장된다.

자식 프로세스에 특정 환경변수만 넘기거나 기존 환경을 덮어써야 한다면 env 인자에 딕셔너리를 넘긴다. 이때 os.environ을 통째로 복사한 뒤 필요한 값만 덮어쓰지 않으면 PATH 같은 필수 변수가 사라져 명령을 찾지 못하는 문제가 생길 수 있다.

import os

env = os.environ.copy()
env["MY_FLAG"] = "1"
subprocess.run(["some-tool"], env=env)

마무리

대부분의 상황에서는 subprocess.run(..., capture_output=True, text=True, check=True) 조합만으로 충분하다. 셸 문법이 꼭 필요한 게 아니라면 shell=True를 피하고 인자를 리스트로 넘기는 것이 안전하고, 실시간 로그 처리처럼 명령이 끝나기 전에 출력을 소비해야 하는 경우에만 Popen으로 내려가면 된다.

참고

Tags: , , , , , , ,

답글 남기기