troubleshooting2024-07-10·8·289/348

Python subprocess 호출 에러 처리

How to properly handle subprocess errors in Python — capturing stderr, dealing with timeouts, and managing zombie processes.

Python subprocess 호출 에러 처리

Introduction

Python's subprocess module is the standard way to run external commands, but it has enough gotchas to fill a book. The most common issues are swallowed errors (where the subprocess fails silently), encoding problems, zombie processes, and timeout handling. These problems are especially dangerous because they often fail silently, leaving your application in an inconsistent state.

I dealt with a critical subprocess error while building a deployment automation tool in Lisbon. The tool ran shell commands to deploy Docker containers, but when a command failed, the error was silently ignored because the return code was not checked. The deployment appeared to succeed even though containers were not running.

Environment

Python 3.12.3
Linux 6.5.0

Problem

Error 1: Command fails silently

import subprocess

# This does NOT raise an exception on failure
result = subprocess.run(
    ["docker", "compose", "up", "-d"],
    capture_output=True,
    text=True
)
# result.returncode might be non-zero, but you never checked
print("Deployment started!")  # Always prints, even if command failed

Error 2: CalledProcessError not caught

import subprocess

try:
    result = subprocess.run(
        ["git", "push", "origin", "main"],
        check=True,  # Raises CalledProcessError on non-zero exit
        capture_output=True,
        text=True
    )
except subprocess.CalledProcessError as e:
    print(f"Command failed: {e.returncode}")
    print(f"Output: {e.output}")
    print(f"Error: {e.stderr}")
Command failed: 1
Output: 
error: failed to push some refs to 'origin'

Error 3: TimeoutExpired exception

import subprocess

try:
    result = subprocess.run(
        ["long_running_command"],
        timeout=10,  # 10 second timeout
        capture_output=True
    )
except subprocess.TimeoutExpired as e:
    print(f"Command timed out after {e.timeout} seconds")
    print(f"Partial output: {e.output}")

Error 4: UnicodeDecodeError with subprocess output

import subprocess

result = subprocess.run(
    ["ls", "-la"],
    capture_output=True
)

# This might fail if output contains non-UTF-8 bytes
text = result.stdout.decode("utf-8")
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xc0 in position 0

Analysis

Subprocess errors have several root causes.

Root Cause 1: Not checking return codes. subprocess.run() does not raise exceptions by default. A non-zero return code indicates failure, but you must check it manually or use check=True.

Root Cause 2: Not capturing stderr. When stderr is not captured, error messages go to the parent process's stderr and may be lost in log aggregation.

Root Cause 3: Blocking indefinitely. Without a timeout, subprocess.run() blocks until the command completes. If the command hangs, your application hangs.

Root Cause 4: Zombie processes. When using subprocess.Popen() directly, you must properly wait for and clean up child processes.

Solution

Fix 1: Always use check=True and handle exceptions

import subprocess

def run_command(cmd, description="Command", timeout=60):
    """Run a subprocess with proper error handling."""
    try:
        result = subprocess.run(
            cmd,
            check=True,
            capture_output=True,
            text=True,
            timeout=timeout
        )
        return result.stdout
    except subprocess.CalledProcessError as e:
        print(f"{description} failed with code {e.returncode}")
        print(f"stderr: {e.stderr}")
        raise
    except subprocess.TimeoutExpired:
        print(f"{description} timed out after {timeout} seconds")
        raise

# Usage
try:
    output = run_command(
        ["docker", "compose", "up", "-d"],
        description="Docker deployment",
        timeout=120
    )
    print(f"Success: {output}")
except subprocess.CalledProcessError:
    print("Deployment failed")

Fix 2: Handle encoding issues

import subprocess

def run_command_safe(cmd, encoding="utf-8", timeout=60):
    """Run command with safe encoding handling."""
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            timeout=timeout
        )
        
        # Decode with error handling
        stdout = result.stdout.decode(encoding, errors="replace")
        stderr = result.stderr.decode(encoding, errors="replace")
        
        if result.returncode != 0:
            raise subprocess.CalledProcessError(
                result.returncode, cmd, result.stdout, result.stderr
            )
        
        return stdout
    except subprocess.TimeoutExpired as e:
        # Kill the process and get partial output
        if e.stdout:
            partial = e.stdout.decode(encoding, errors="replace")
            print(f"Partial output: {partial}")
        raise

output = run_command_safe(["ls", "-la"])

Fix 3: Use communicate() for complex I/O

import subprocess

def run_with_input(cmd, input_data, timeout=30):
    """Run command with stdin input."""
    process = subprocess.Popen(
        cmd,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )
    
    try:
        stdout, stderr = process.communicate(
            input=input_data.encode(),
            timeout=timeout
        )
        return {
            "returncode": process.returncode,
            "stdout": stdout.decode("utf-8", errors="replace"),
            "stderr": stderr.decode("utf-8", errors="replace")
        }
    except subprocess.TimeoutExpired:
        process.kill()
        process.wait()
        raise

result = run_with_input(
    ["python3", "-c", "print(input())"],
    input_data="hello",
    timeout=5
)

Fix 4: Handle zombie processes with Popen

import subprocess
import signal

def run_background(cmd):
    """Run command in background with proper cleanup."""
    process = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        start_new_session=True  # Detach from parent
    )
    
    try:
        stdout, stderr = process.communicate(timeout=30)
        return process.returncode, stdout, stderr
    except subprocess.TimeoutExpired:
        # Send SIGTERM first, then SIGKILL
        process.terminate()
        try:
            process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            process.kill()
            process.wait()
        raise
    finally:
        # Ensure process is cleaned up
        if process.poll() is None:
            process.kill()
            process.wait()

Fix 5: Use shlex for safe command construction

import subprocess
import shlex

# Never do this (shell injection risk):
# subprocess.run(f"echo {user_input}", shell=True)

# Do this instead:
user_input = "hello world"
cmd = ["echo", user_input]
result = subprocess.run(cmd, capture_output=True, text=True)
print(result.stdout)

# Or with shell=True and shlex
cmd_string = f"echo {shlex.quote(user_input)}"
result = subprocess.run(cmd_string, shell=True, capture_output=True, text=True)

Lessons Learned

  • Always use check=True unless you specifically need to handle non-zero exit codes manually.
  • Always set a timeout to prevent subprocess calls from blocking indefinitely.
  • Use errors="replace" when decoding subprocess output to handle unexpected bytes.
  • Use start_new_session=True for background processes to prevent zombie processes.
  • Never use shell=True with untrusted input — use shlex.quote() or pass arguments as a list.

This blog does not accept any external sponsorships, affiliate marketing, or ad revenue.