subprocess — Spawning Processes#

What it is#

subprocess is Python’s standard library for spawning new processes, sending data through their stdin, and reading their stdout, stderr, and exit code. It has been the recommended approach since Python 3.5 (PEP 324, refined by PEP 446), superseding the older os.system, os.popen, and the commands module. Reach for subprocess whenever you need to shell out to another binary (git, ffmpeg, aws, psql) from a Python script — and reach for subprocess.run first, dropping down to Popen only when you need streaming I/O or finer control.

Install#

subprocess is part of the Python standard library and requires no installation. Verify it loads:

python -c "import subprocess; print(subprocess.run(['echo', 'ok'], capture_output=True, text=True).stdout)"

Output:

ok

Mental model#

A subprocess.run call wraps the entire lifecycle of one external command: fork-exec a child, optionally feed it stdin, wait for it to exit, and return a CompletedProcess with returncode, stdout, and stderr. Popen is the lower-level object — run is built on top of it — and is what you use when you need to communicate while the child is still running.

import subprocess

result = subprocess.run(
    ["ls", "-1", "/home/alice"],
    capture_output=True,
    text=True,
    check=True,
)
print("returncode:", result.returncode)
print("stdout:", result.stdout.strip())

Output:

returncode: 0
stdout: Documents
Downloads
projects

subprocess.run — the preferred entry point#

subprocess.run(args, ...) runs args to completion and returns a CompletedProcess. It blocks until the child exits, handles cleanup automatically, and is documented as the right call for “the majority of use cases.” Pass args as a list of strings, not a single string — that’s the safe form that avoids the shell.

import subprocess

cp = subprocess.run(["git", "status", "--short"], capture_output=True, text=True)
print(cp.returncode)
print(cp.stdout)

Output:

0
 M README.md
?? notes.txt

Essential keyword arguments#

subprocess.run accepts a few dozen keyword arguments but most calls use the same handful. The table below covers everything you’ll reach for day-to-day.

capture_output=True, text=True, check=True#

The single most useful triple. Together they say: “run this command, give me its output as a string, and raise if it failed.” Use them as the default for any command whose output you care about.

import subprocess

cp = subprocess.run(
    ["uname", "-a"],
    capture_output=True,
    text=True,
    check=True,
)
print(cp.stdout.strip())

Output:

Linux myhost 6.5.0-15-generic #15-Ubuntu SMP Tue Jan  9 19:11:25 UTC 2026 x86_64 GNU/Linux

Exit codes and CalledProcessError#

check=True raises CalledProcessError on non-zero exit. The exception carries returncode, cmd, stdout, stderr, making it ideal for “fail fast” pipelines. Without check=True, inspect cp.returncode yourself.

import subprocess

try:
    subprocess.run(["ls", "/no/such/path"], capture_output=True, text=True, check=True)
except subprocess.CalledProcessError as e:
    print("rc:", e.returncode)
    print("cmd:", e.cmd)
    print("stderr:", e.stderr.strip())

Output:

rc: 2
cmd: ['ls', '/no/such/path']
stderr: ls: cannot access '/no/such/path': No such file or directory

Passing input via stdin#

input= writes a string (or bytes) to the child’s stdin, then closes the stream. This is how you feed a heredoc-style payload to a command without temp files. Combine with capture_output=True, text=True to round-trip text.

import subprocess

cp = subprocess.run(
    ["wc", "-w"],
    input="alice bob charlie\nalice charlie\n",
    capture_output=True,
    text=True,
    check=True,
)
print("word count:", cp.stdout.strip())

Output:

word count: 5

timeout= and process killing#

timeout= kills the child after N seconds and raises TimeoutExpired. The exception carries any output captured before the kill. Always wrap network-bound commands in a timeout — there’s no other way to bound their runtime.

import subprocess

try:
    subprocess.run(["sleep", "10"], timeout=1, capture_output=True, text=True)
except subprocess.TimeoutExpired as e:
    print(f"timed out after {e.timeout}s; killed {e.cmd}")

Output:

timed out after 1s; killed ['sleep', '10']

Overriding cwd and env#

cwd= sets the child’s working directory; env= replaces (not extends) the child’s environment. To extend, copy os.environ first. Both are essential when scripting against git (which is cwd-sensitive) or when injecting credentials without leaking them into your own process.

import os
import subprocess

env = os.environ.copy()
env["GIT_PAGER"] = "cat"             # disable pager
env["LANG"] = "C"                    # force English error messages
env["GIT_AUTHOR_NAME"] = "Alice Dev"  # only for this child

cp = subprocess.run(
    ["git", "log", "-1", "--format=%an %ae"],
    cwd="/home/alice/projects/demo",
    env=env,
    capture_output=True,
    text=True,
    check=True,
)
print(cp.stdout.strip())

Output:

Alice Dev alice@example.com

[!WARNING] env={} clears the child’s environment entirely — including PATH. The child won’t find executables by name. If you only need to add a variable, copy os.environ first.

shell=True and why to avoid it#

When shell=True, the first argument is passed verbatim to /bin/sh -c (or cmd.exe /c). This enables shell features (globbing, pipes, redirection, environment expansion) but opens you to command injection if any part of the string comes from user input. The default shell=False exec’s the binary directly and is safe by construction.

import subprocess

# DON'T (vulnerable if `user_arg` came from outside):
subprocess.run(f"ls {user_arg}", shell=True)

# DO:
subprocess.run(["ls", user_arg])

Output: (none — security note)

[!WARNING] Never interpolate untrusted strings into a shell=True command. Use a list of args, or escape with shlex.quote if you absolutely must.

shlex.quote for safe shell strings#

When you genuinely need a single shell command string (logging it, writing a .sh file, building an SSH command line), shlex.quote wraps each argument in single quotes so the shell sees it as one literal token. shlex.join does the whole list at once.

import shlex
import subprocess

paths = ["/home/alice/My Docs", "/tmp/notes; rm -rf /"]
safe = shlex.join(["ls", "-la", *paths])
print(safe)

# Pipe through `ssh` to run on a remote host
subprocess.run(["ssh", "alicedev@myhost.local", safe], check=True)

Output:

ls -la '/home/alice/My Docs' '/tmp/notes; rm -rf /'

Popen — the lower-level API#

Popen constructs the child process and returns immediately, without waiting. You can write to proc.stdin, read from proc.stdout / proc.stderr, and call proc.wait() / proc.communicate() / proc.terminate() / proc.kill() as needed. Use it when you need streaming, multiple concurrent children, or pipelines.

import subprocess

proc = subprocess.Popen(
    ["ping", "-c", "3", "127.0.0.1"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True,
)
stdout, _ = proc.communicate(timeout=10)
print("returncode:", proc.returncode)
print(stdout.splitlines()[0])

Output:

returncode: 0
PING 127.0.0.1 (127.0.0.1) 56(84) bytes of data.

Streaming stdout line by line#

For long-running commands, you want to react to output as it arrives, not after the child exits. Read proc.stdout like a file: each iteration yields the next line. This is the right way to tail a tail -f, docker logs, or npm run build.

import subprocess

with subprocess.Popen(
    ["ping", "-c", "4", "8.8.8.8"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True,
    bufsize=1,                # line-buffered
) as proc:
    for line in proc.stdout:
        print(">", line.rstrip())
    if proc.wait() != 0:
        raise subprocess.CalledProcessError(proc.returncode, proc.args)

Output:

> PING 8.8.8.8 (8.8.8.8) 56(84) bytes of data.
> 64 bytes from 8.8.8.8: icmp_seq=1 ttl=117 time=8.42 ms
> 64 bytes from 8.8.8.8: icmp_seq=2 ttl=117 time=7.91 ms
> 64 bytes from 8.8.8.8: icmp_seq=3 ttl=117 time=8.15 ms
> 64 bytes from 8.8.8.8: icmp_seq=4 ttl=117 time=8.03 ms

[!TIP] Always wrap Popen in a with block. The context manager closes pipes and waits for the child, preventing zombies and file-descriptor leaks.

Piping one process into another#

To replicate ps aux | grep python, wire the stdout of the first Popen into the stdin of the second. Close the upstream stdout after handing it off so the downstream sees EOF when the source finishes.

import subprocess

ps = subprocess.Popen(["ps", "aux"], stdout=subprocess.PIPE)
grep = subprocess.Popen(
    ["grep", "python"],
    stdin=ps.stdout,
    stdout=subprocess.PIPE,
    text=True,
)
ps.stdout.close()                    # let grep see EOF
out, _ = grep.communicate(timeout=5)
print(out.splitlines()[0])

Output:

alicedev 12345  0.1  0.4  74132 33872 ?  Sl  14:22  0:00 python /home/alice/app.py

Pipelines without two Popens#

If you don’t need the streaming, the simpler approach is to feed the first command’s captured stdout into the second via input=. Two blocking run() calls — easier to read and debug.

import subprocess

ps = subprocess.run(["ps", "aux"], capture_output=True, text=True, check=True)
grep = subprocess.run(
    ["grep", "python"],
    input=ps.stdout,
    capture_output=True,
    text=True,
)
print(grep.stdout.splitlines()[0])

Output:

alicedev 12345  0.1  0.4  74132 33872 ?  Sl  14:22  0:00 python /home/alice/app.py

Redirecting to a file#

stdout= and stderr= accept any file object (or file descriptor int). Open a file in write mode, pass the handle, and the child writes directly to disk without Python buffering its output. Combine with subprocess.STDOUT to merge stderr into stdout.

import subprocess
from pathlib import Path

log = Path("/home/alice/logs/build.log")
log.parent.mkdir(parents=True, exist_ok=True)
with log.open("w", encoding="utf-8") as f:
    subprocess.run(
        ["pip", "install", "-r", "requirements.txt"],
        stdout=f,
        stderr=subprocess.STDOUT,
        check=True,
    )
print("wrote", log)

Output:

wrote /home/alice/logs/build.log

Discarding output#

Set stdout=subprocess.DEVNULL (and/or stderr=DEVNULL) to throw away output cleanly. Faster and simpler than > /dev/null redirection through a shell.

import subprocess

# Run a noisy command and ignore everything but the exit code
rc = subprocess.run(
    ["pre-commit", "run", "--all-files"],
    stdout=subprocess.DEVNULL,
    stderr=subprocess.DEVNULL,
).returncode
print("hooks pass" if rc == 0 else f"hooks fail (rc={rc})")

Output:

hooks pass

Sending signals#

proc.terminate() sends SIGTERM (graceful stop). proc.kill() sends SIGKILL (immediate). proc.send_signal(sig) sends any signal. Always give the process a chance to clean up before escalating.

import signal
import subprocess
import time

proc = subprocess.Popen(["sleep", "30"])
time.sleep(1)
proc.send_signal(signal.SIGTERM)
try:
    proc.wait(timeout=2)
except subprocess.TimeoutExpired:
    proc.kill()
    proc.wait()
print("returncode:", proc.returncode)

Output:

returncode: -15

[!NOTE] A negative returncode means the child was terminated by signal N (-15 = killed by SIGTERM). On Windows, terminate() and kill() both call TerminateProcess and the returncode is 1.

Handling Ctrl-C cleanly#

When the user hits Ctrl-C, both your Python process and any child sharing the same terminal receive SIGINT. Wrap your Popen in a try/except KeyboardInterrupt to kill the child and re-raise. Without this, the child keeps running after your script exits.

import subprocess

try:
    with subprocess.Popen(["ffmpeg", "-i", "in.mp4", "out.mkv"]) as proc:
        proc.wait()
except KeyboardInterrupt:
    proc.terminate()
    try:
        proc.wait(timeout=5)
    except subprocess.TimeoutExpired:
        proc.kill()
    print("\nffmpeg interrupted; cleaned up")

Output:

^C
ffmpeg interrupted; cleaned up

Async subprocesses with asyncio#

asyncio.create_subprocess_exec is the async cousin of Popen — fire dozens of commands concurrently without blocking the event loop. Use it inside async code (FastAPI handlers, async CLIs) where blocking subprocess.run would stall the whole server.

import asyncio

async def run(*args: str) -> tuple[int, str]:
    proc = await asyncio.create_subprocess_exec(
        *args,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.STDOUT,
    )
    out, _ = await proc.communicate()
    return proc.returncode, out.decode()

async def main():
    rcs = await asyncio.gather(
        run("git", "rev-parse", "HEAD"),
        run("git", "branch", "--show-current"),
    )
    for rc, out in rcs:
        print(rc, out.strip())

asyncio.run(main())

Output:

0 7c4f3a2b9d1e5f6c8a0b2d4e6f8a0c2e4f6a8c0e
0 main

Windows note — cmd.exe vs PowerShell#

On Windows, subprocess calls CreateProcess directly. Built-in shell commands (dir, copy, echo) are not exe files — they only exist inside cmd.exe. To use them you need shell=True (which spawns cmd.exe). For PowerShell, invoke powershell.exe -NoProfile -Command ... or pwsh -Command ... directly.

import subprocess

# WRONG on Windows — `dir` is not an exe
# subprocess.run(["dir"])  # FileNotFoundError

# RIGHT: use a real exe
subprocess.run(["powershell", "-NoProfile", "-Command", "Get-ChildItem"], check=True)

Output: (none — PowerShell prints directory listing)

subprocess vs alternatives#

For most tasks subprocess.run is the right answer. Where Python alone can do the job — file listing, regex matching, JSON parsing — prefer pure Python; spawning processes is 10–100× slower than the in-process call.

Common pitfalls#

1. Passing a single string with shell=False raises FileNotFoundError. ["ls -la"] looks for an executable literally named ls -la. Use ["ls", "-la"] or set shell=True.
2. shell=True with untrusted input is a remote-code-execution bug. Use a list of args or shlex.quote.
3. env={} clears PATH. Copy os.environ first.
4. Forgetting text=True gives you bytes. Decoding manually with .decode() works but text=True is cleaner and lets you pick encoding= and errors="replace".
5. Popen without with leaks pipes and creates zombies. Always use a context manager.
6. Deadlock on large output — using proc.stdout.read() while the child also fills stderr blocks both sides forever. Use communicate(), or redirect one stream to DEVNULL/STDOUT.
7. Timeout without kill() + wait() — TimeoutExpired is raised but the child may still be alive. Always proc.kill(); proc.wait() in the except block.
8. Globbing in args doesn’t work: ["ls", "*.txt"] looks for a literal *.txt. Either glob in Python (Path.glob) or use shell=True deliberately.
9. Reading line-by-line without bufsize=1 can hang because the child’s stdout is fully buffered when not a TTY. Set bufsize=1 (line-buffered) for streaming.

Real-world recipes#

Run a long command, stream to a log, kill on Ctrl-C#

The canonical pattern for wrapping a long-running build/encode/test command — capture output to a log file, mirror to stdout, and shut down cleanly when interrupted.

import signal
import subprocess
import sys
from pathlib import Path

def run_logged(args: list[str], log_path: Path) -> int:
    log_path.parent.mkdir(parents=True, exist_ok=True)
    with subprocess.Popen(
        args,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1,
    ) as proc, log_path.open("w", encoding="utf-8") as log:
        try:
            for line in proc.stdout:
                sys.stdout.write(line)
                log.write(line)
            return proc.wait()
        except KeyboardInterrupt:
            proc.send_signal(signal.SIGINT)
            try:
                return proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                proc.kill()
                return proc.wait()

if __name__ == "__main__":
    rc = run_logged(
        ["ffmpeg", "-i", "/home/alice/in.mp4", "/home/alice/out.mkv"],
        Path("/home/alice/logs/ffmpeg.log"),
    )
    sys.exit(rc)

Output:

ffmpeg version 6.1 Copyright (c) 2000-2026 the FFmpeg developers
...
frame=  240 fps=120 q=27.0 size=    1024kB time=00:00:08.00 bitrate=1048.5kbits/s
^C

Get the current git commit SHA#

A common boilerplate for stamping build artifacts with the source revision. Fall back to unknown if the command isn’t available (running in a release tarball without .git).

import subprocess

def git_sha() -> str:
    try:
        return subprocess.run(
            ["git", "rev-parse", "--short", "HEAD"],
            capture_output=True,
            text=True,
            check=True,
            cwd="/home/alice/projects/demo",
        ).stdout.strip()
    except (subprocess.CalledProcessError, FileNotFoundError):
        return "unknown"

print("build:", git_sha())

Output:

build: 7c4f3a2

Run a command on a remote host via SSH#

Combine subprocess.run with ssh and shlex.join to invoke arbitrary commands on another machine. Pass -o BatchMode=yes to fail fast if interactive auth would be needed.

import shlex
import subprocess

def remote(host: str, *args: str) -> str:
    cmd = ["ssh", "-o", "BatchMode=yes", host, shlex.join(args)]
    return subprocess.run(
        cmd, capture_output=True, text=True, check=True, timeout=30
    ).stdout

print(remote("alicedev@myhost.local", "df", "-h", "/"))

Output:

Filesystem      Size  Used Avail Use% Mounted on
/dev/sda1       100G   45G   55G  45% /

Tail a file with tail -f and react per line#

tail -f keeps printing as new lines arrive. Spawn it with Popen, iterate over proc.stdout, and process each line — a one-page implementation of a structured log monitor.

import re
import subprocess

ERROR_RE = re.compile(r"\bERROR\b")

with subprocess.Popen(
    ["tail", "-F", "/home/alice/logs/app.log"],
    stdout=subprocess.PIPE,
    text=True,
    bufsize=1,
) as proc:
    for line in proc.stdout:
        if ERROR_RE.search(line):
            print("ALERT:", line.rstrip())

Output:

ALERT: 2026-05-25 14:55:11 ERROR connection refused to db.internal
ALERT: 2026-05-25 14:55:14 ERROR timeout after 30s

Cap output and never block the event loop (asyncio)#

In an async server, calling subprocess.run blocks the thread. Use asyncio.create_subprocess_exec plus asyncio.wait_for to enforce a timeout without stalling other requests.

import asyncio
from asyncio.subprocess import PIPE

async def run_bounded(args: list[str], timeout: float = 10.0) -> str:
    proc = await asyncio.create_subprocess_exec(*args, stdout=PIPE, stderr=PIPE)
    try:
        out, err = await asyncio.wait_for(proc.communicate(), timeout=timeout)
    except asyncio.TimeoutError:
        proc.kill()
        await proc.wait()
        raise
    if proc.returncode:
        raise RuntimeError(f"rc={proc.returncode}: {err.decode()}")
    return out.decode()

async def main():
    print(await run_bounded(["uname", "-r"]))

asyncio.run(main())

Output:

6.5.0-15-generic