Python Subprocess — Running External Commands

The Story That Explains subprocess 💈subprocess समझाने वाली कहानी 💈一个讲清 subprocess 的故事 💈

Every operating system runs work as processes. When your script needs to run ls -la, convert a video with ffmpeg, call git status or start any other program, it uses subprocess to start a child process, give it the command, and talk to it through three streams: stdin, stdout and stderr.

हर operating system काम को processes के रूप में चलाता है। जब आपकी script को ls -la चलाना हो, ffmpeg से video बदलना हो, git status बुलाना हो या कोई और program शुरू करना हो, तो वह subprocess से child process शुरू करती है, उसे command देती है, और तीन streams से उससे बात करती है: stdin, stdout और stderr।

每个操作系统都以 进程 的形式运行任务。当你的脚本需要运行 ls -la、用 ffmpeg 转换视频、调用 git status 或启动其他任何程序时，它会用 subprocess 启动一个子进程，把命令交给它，并通过三个流与它通信：stdin、stdout 和 stderr。

The Foundation — What Actually Happens 🧱नींव — असल में क्या होता है 🧱基础 — 实际发生了什么 🧱

When you call subprocess.run(["ls", "-la"]), Python asks the operating system to create a new process, connect its streams, run the program and report back how it ended:

जब आप subprocess.run(["ls", "-la"]) बुलाते हैं, तो Python operating system से नया process बनाने, उसकी streams जोड़ने, program चलाने और यह बताने को कहता है कि वह कैसे ख़त्म हुआ:

当你调用 subprocess.run(["ls", "-la"]) 时，Python 请求操作系统创建一个新进程、连接它的数据流、运行程序，并报告它是如何结束的：

subprocess.run — The Everyday Tool 🛠️subprocess.run — रोज़ का औज़ार 🛠️subprocess.run — 日常工具 🛠️

subprocess.run() is your default. It runs a command, waits for it to finish and returns a CompletedProcess object with the exit code and (if you ask) the captured output. Reach for it first.

subprocess.run() आपका default है। यह command चलाता है, उसके ख़त्म होने का इंतज़ार करता है, और exit code और (माँगने पर) पकड़ा गया output लिए एक CompletedProcess object लौटाता है। पहले इसी को चुनिए।

subprocess.run() 是你的默认选择。它运行命令、等待 它结束，然后返回一个 CompletedProcess 对象，里面有退出码以及（如果你要求的话）捕获的输出。首先用它。

import subprocess

# Simplest possible call - just run it, wait, done
result = subprocess.run(["echo", "Hello, Mohit!"])
print(f"return code: {result.returncode}")

Notice that Hello, Mohit! was printed by the child, straight to the screen — Python never saw that text. To get the output into a variable, capture it:

ध्यान दीजिए, Hello, Mohit! child ने सीधे screen पर print किया — Python ने वह text देखा ही नहीं। Output को variable में लेने के लिए उसे capture कीजिए:

注意，Hello, Mohit! 是 子进程 直接打印到屏幕上的 — Python 根本没看到这段文字。要把输出存进变量，就要捕获它：

Capturing Output — capture_output + textOutput पकड़ना — capture_output + text捕获输出 — capture_output + text

import subprocess

result = subprocess.run(
    ["ls", "-la", "reports"],
    capture_output=True,       # capture stdout + stderr
    text=True                  # decode bytes to str
)

print("--- STDOUT ---")
print(result.stdout)
print("--- STDERR ---")
print(result.stderr or "(none)")
print(f"exit code: {result.returncode}")

Handling Errors — check=True and returncode 🚦Errors सँभालना — check=True और returncode 🚦处理错误 — check=True 与 returncode 🚦

Every process ends with an exit code: 0 for success, anything else for failure. By default subprocess.run does not raise an error when a command fails — you must check yourself, or pass check=True.

हर process एक exit code के साथ ख़त्म होता है: सफलता के लिए 0, असफलता के लिए कुछ और। Default में command fail होने पर subprocess.run error raise नहीं करता — आपको ख़ुद जाँचना होगा, या check=True देना होगा।

每个进程结束时都有一个 退出码：0 表示成功，其他值表示失败。默认情况下，命令失败时 subprocess.run 不会 抛出异常 — 你必须自己检查，或者传入 check=True。

import subprocess

# No check=True: the failure is easy to miss
result = subprocess.run(["cat", "/does/not/exist"], capture_output=True, text=True)
print("script keeps going... returncode =", result.returncode)

import subprocess

# Command that will fail - no such file
try:
    result = subprocess.run(
        ["cat", "/does/not/exist"],
        capture_output=True,
        text=True,
        check=True               # raise on non-zero exit
    )
except subprocess.CalledProcessError as e:
    print(f"Command failed with code {e.returncode}")
    print(f"stderr: {e.stderr.strip()}")
    print(f"cmd:    {e.cmd}")

Timeouts — Never Wait Forever ⏱️Timeouts — कभी हमेशा इंतज़ार नहीं ⏱️超时 — 永远不要无限等待 ⏱️

A stuck process will freeze your script for ever. Always set a timeout when you are not sure how long a command will take. Here a small command prints one "tick" per second; we give it 2.5 seconds:

अटका हुआ process आपकी script को हमेशा के लिए रोक देगा। जब पक्का न हो कि command कितना समय लेगा, तो हमेशा timeout दीजिए। यहाँ एक छोटा command हर second एक "tick" print करता है; हम उसे 2.5 seconds देते हैं:

卡住的进程会让你的脚本永远停在那里。只要不确定命令要运行多久，就一定要设置 timeout。这里一个小命令每秒打印一个"tick"，我们只给它 2.5 秒：

import subprocess

try:
    result = subprocess.run(
        ["bash", "-c", "for i in 1 2 3 4 5; do echo tick $i; sleep 1; done"],
        capture_output=True,
        text=True,
        timeout=2.5              # seconds - kill after this
    )
    print(result.stdout)
except subprocess.TimeoutExpired as e:
    print(f"Timed out after {e.timeout}s")
    # e.stdout is ALWAYS bytes here, even with text=True
    partial = e.stdout.decode() if e.stdout else "none"
    print("Partial output:")
    print(partial, end="")

🧪 Simulator: run() Lab🧪 Simulator: run() Lab🧪 模拟器：run() 实验室

Pick a command and switch the four options on and off. Watch where stdout and stderr go, what type r.stdout becomes, and what appears on the screen. All 64 combinations were run for real; the time is the real time each one took.

एक command चुनिए और चारों options चालू-बंद कीजिए। देखिए stdout और stderr कहाँ जाते हैं, r.stdout किस type का बनता है, और screen पर क्या आता है। सभी 64 combinations असल में चलाए गए; समय वही असली समय है जो हर एक ने लिया।

选一个命令，打开或关闭四个选项。观察 stdout 和 stderr 流向哪里、r.stdout 变成什么类型，以及屏幕上出现了什么。全部 64 种组合都真实运行过；显示的时间就是每次运行的真实耗时。

subprocess.run — Key Arguments 🎛️subprocess.run — मुख्य Arguments 🎛️subprocess.run — 关键参数 🎛️

Passing stdin — sending data to the childstdin देना — child को data भेजना传入 stdin — 向子进程发送数据

import subprocess

# Sort a list of names by piping stdin
names = "charlie\nalice\nbob\ndave"

result = subprocess.run(
    ["sort"],
    input=names,                 # fed to child's stdin
    capture_output=True,
    text=True,
    check=True
)
print(result.stdout)

Custom env and cwdअपना env और cwd自定义 env 与 cwd

import subprocess, os

# Run a command in a specific directory with a modified env
custom_env = os.environ.copy()
custom_env["MY_API_KEY"] = "secret-token-123"
custom_env["PYTHONUNBUFFERED"] = "1"

result = subprocess.run(
    ["bash", "-c", "echo $MY_API_KEY; pwd"],
    cwd="/tmp",
    env=custom_env,
    capture_output=True,
    text=True
)
print(result.stdout)

Popen — Low-Level Control for Streaming 📡Popen — Streaming के लिए Low-Level नियंत्रण 📡Popen — 用于流式处理的底层控制 📡

subprocess.run is built on top of subprocess.Popen. Use Popen directly when you need to:

subprocess.run, subprocess.Popen के ऊपर बना है। Popen सीधे तब इस्तेमाल कीजिए जब आपको चाहिए:

subprocess.run 是在 subprocess.Popen 之上构建的。在以下情况下直接使用 Popen：

Streaming stdout line by linestdout को line दर line पढ़ना逐行读取 stdout

import subprocess

# A task that prints one line per second
p = subprocess.Popen(
    ["bash", "-c",
     "for i in 1 2 3 4 5; do echo 'line '$i; sleep 1; done"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,        # merge stderr into stdout
    text=True,
)

# Read one line at a time as it's produced
for line in p.stdout:
    print(f"[live] {line.rstrip()}")

exit_code = p.wait()
print(f"\ndone, exit={exit_code}")

📡 Simulator: Streaming Lab📡 Simulator: Streaming Lab📡 模拟器：流式输出实验室

The child prints one line every 0.5 seconds. Pick what kind of child it is and the bufsize, then press Play. The top track shows when the child printed; the bottom track shows when Python received each line. All times were measured in real runs (rounded to 0.5 s).

Child हर 0.5 second में एक line print करता है। Child का प्रकार और bufsize चुनिए, फिर Play दबाइए। ऊपर वाला track दिखाता है कि child ने कब print किया; नीचे वाला दिखाता है कि Python को हर line कब मिली। सारे समय असली runs में नापे गए (0.5 s तक गोल किए)।

子进程每 0.5 秒打印一行。选择子进程的类型和 bufsize，然后点击播放。上面的轨道显示子进程何时 打印；下面的轨道显示 Python 何时 收到 每一行。所有时间都来自真实运行的测量（四舍五入到 0.5 秒）。

Pipes — Chaining Commands Like the Shell 🔗Pipes — Shell की तरह Commands जोड़ना 🔗管道 — 像 shell 一样串联命令 🔗

The shell command ps aux | grep python | wc -l chains three programs: the output of ps feeds grep, and the output of grep feeds wc. In Python you build the same chain by connecting one process's stdout to the next one's stdin.

Shell command ps aux | grep python | wc -l तीन programs जोड़ता है: ps का output grep में जाता है, और grep का output wc में। Python में आप एक process का stdout अगले के stdin से जोड़कर यही chain बनाते हैं।

shell 命令 ps aux | grep python | wc -l 串联了三个程序：ps 的输出喂给 grep，grep 的输出再喂给 wc。在 Python 中，把一个进程的 stdout 接到下一个进程的 stdin，就能搭出同样的链。

Building the pipeline in PythonPython में pipeline बनाना在 Python 中搭建管道

import subprocess

# Equivalent of:  ps aux | grep python | wc -l
p1 = subprocess.Popen(
    ["ps", "aux"],
    stdout=subprocess.PIPE
)
p2 = subprocess.Popen(
    ["grep", "python"],
    stdin=p1.stdout,                 # feed p1's output
    stdout=subprocess.PIPE
)
p3 = subprocess.Popen(
    ["wc", "-l"],
    stdin=p2.stdout,                 # feed p2's output
    stdout=subprocess.PIPE,
    text=True
)

# Close upstream stdouts in the parent so SIGPIPE works
p1.stdout.close()
p2.stdout.close()

count = p3.communicate()[0].strip()
print(f"python processes: {count}")

🔗 Simulator: Pipeline Flow🔗 Simulator: Pipeline Flow🔗 模拟器：管道流程

Here ps aux is replaced by a fixed sample list, so everyone gets the same result. The last line is the grep process itself — just as it would appear in a real ps list. Pick a word and watch each line flow through grep into wc -l. The counts were produced by a real Popen pipeline.

यहाँ ps aux की जगह एक तय sample सूची है, ताकि सबको एक ही नतीजा मिले। आख़िरी line ख़ुद grep process है — जैसे असली ps सूची में दिखता। एक शब्द चुनिए और हर line को grep से होकर wc -l तक जाते देखिए। गिनतियाँ असली Popen pipeline से बनी हैं।

这里用一份固定的示例列表代替 ps aux，让每个人得到相同的结果。最后一行就是 grep 进程本身 — 就像在真实的 ps 列表中那样。选一个词，看每一行如何流经 grep 进入 wc -l。计数是由真实的 Popen 管道得出的。

The shortcut — two run() callsछोटा रास्ता — दो run() calls捷径 — 两次 run() 调用

import subprocess

# Disk usage of the root filesystem: df -h | grep " /$"
result = subprocess.run(
    ["grep", " /$"],
    input=subprocess.run(["df", "-h"], capture_output=True, text=True).stdout,
    capture_output=True,
    text=True
)
print(result.stdout)

communicate() — Send and Receive Safely 📨communicate() — सुरक्षित भेजना और पाना 📨communicate() — 安全地发送与接收 📨

When you start a process with Popen and need to send input and then read all the output, use communicate(). It writes the input, closes stdin, reads all of stdout and stderr at the same time, and waits for the child to exit — without deadlocks.

जब आप Popen से process शुरू करें और input भेजकर फिर पूरा output पढ़ना हो, तो communicate() लीजिए। यह input लिखता है, stdin बंद करता है, stdout और stderr दोनों एक साथ पूरे पढ़ता है, और child के exit होने का इंतज़ार करता है — बिना deadlock के।

当你用 Popen 启动进程、需要先发送输入再读取全部输出时，使用 communicate()。它会写入输入、关闭 stdin、同时读取全部 stdout 和 stderr，并等待子进程退出 — 不会死锁。

import subprocess

p = subprocess.Popen(
    ["python3"],                 # Python reading a script from stdin
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

script = """
import math
print('sqrt(2) =', math.sqrt(2))
print('pi     =', math.pi)
"""

stdout, stderr = p.communicate(input=script, timeout=5)
print("--- output ---")
print(stdout)
print(f"exit: {p.returncode}")

🔒 Simulator: Deadlock Lab🔒 Simulator: Deadlock Lab🔒 模拟器：死锁实验室

The child writes a block of text. Choose its size and how the parent waits. On our Linux test machine the pipe holds exactly 64 KB; anything bigger with "wait first" gets stuck, and our 2-second timeout rescues the script. Every result was produced by a real run.

Child text का एक block लिखता है। उसका size और parent के इंतज़ार का तरीक़ा चुनिए। हमारी Linux test machine पर pipe ठीक 64 KB रखती है; "पहले wait" के साथ इससे बड़ा कुछ भी अटक जाता है, और हमारा 2-second timeout script को बचाता है। हर नतीजा असली run से है।

子进程写出一块文本。选择它的大小以及父进程的等待方式。在我们的 Linux 测试机上，管道恰好能容纳 64 KB；用"先 wait"时，任何更大的数据都会卡住，我们设置的 2 秒超时把脚本救了回来。每个结果都来自真实运行。

The shell=True Trap — Injection 🚨shell=True का जाल — Injection 🚨shell=True 陷阱 — 注入 🚨

shell=True runs your command through /bin/sh -c "...". It is convenient, but if any part of the string comes from a user, you have a shell injection hole: the shell treats characters like ;, $( ) and * as instructions.

shell=True आपका command /bin/sh -c "..." से चलाता है। यह सुविधाजनक है, पर अगर string का कोई भी हिस्सा user से आता है, तो आपके पास shell injection का छेद है: shell ;, $( ) और * जैसे characters को निर्देश मानता है।

shell=True 会通过 /bin/sh -c "..." 运行你的命令。它很方便，但只要字符串中有 任何 部分来自用户，你就有了 shell 注入 漏洞：shell 会把 ;、$( )、* 等字符当作指令。

import subprocess

user_input = "nothing.txt; echo HACKED - this command was injected"

print("--- shell=True with an f-string ---")
subprocess.run(f"cat {user_input}", shell=True)

print("--- list form ---")
subprocess.run(["cat", user_input])

One more trap: inputs that look like optionsएक और जाल: options जैसे दिखने वाले inputs另一个陷阱：看起来像选项的输入

import subprocess

user_file = "--version"          # looks like a filename, acts like an option

print("--- list form ---")
r = subprocess.run(["cat", user_file], capture_output=True, text=True)
print(r.stdout.splitlines()[0])

print("--- list form with -- ---")
r = subprocess.run(["cat", "--", user_file], capture_output=True, text=True)
print(r.stderr.strip())

🚨 Simulator: Injection Lab🚨 Simulator: Injection Lab🚨 模拟器：注入实验室

Pick a "user input" and a way to call cat. See exactly which arguments cat receives and what really happens. The arguments were printed by a real helper program, and every output comes from a real run in a folder that holds report.txt and my file.txt.

एक "user input" और cat बुलाने का तरीक़ा चुनिए। देखिए cat को ठीक कौन-से arguments मिलते हैं और असल में क्या होता है। Arguments एक असली helper program ने print किए, और हर output report.txt और my file.txt वाले folder में असली run से है।

选一个"用户输入"和一种调用 cat 的方式。看看 cat 究竟收到了哪些参数，以及实际发生了什么。参数是由一个真实的辅助程序打印的，每个输出都来自在包含 report.txt 和 my file.txt 的文件夹中的真实运行。

Real-World Example — Git Status Dashboard 🗂️असली Example — Git Status Dashboard 🗂️真实案例 — Git 状态面板 🗂️

A small tool that checks several git repositories and reports which ones have uncommitted changes. It uses run, check=True, a timeout, cwd and error handling — the pieces you need in real scripts. For a real run we created four sample folders: api (3 changed files), web (clean, tracking a remote branch), docs (a plain folder, not a repo) and blog (does not exist).

एक छोटा tool जो कई git repositories जाँचता है और बताता है किनमें uncommitted बदलाव हैं। इसमें run, check=True, timeout, cwd और error handling है — असली scripts के ज़रूरी हिस्से। असली run के लिए हमने चार sample folders बनाए: api (3 बदली files), web (साफ़, remote branch को track करता), docs (सादा folder, repo नहीं) और blog (मौजूद ही नहीं)।

一个检查多个 git 仓库、报告哪些有未提交改动的小工具。它用到了 run、check=True、超时、cwd 和错误处理 — 正是真实脚本需要的部件。为了真实运行，我们创建了四个示例文件夹：api（3 个改动的文件）、web（干净，跟踪一个远程分支）、docs（普通文件夹，不是仓库）和 blog（根本不存在）。

import subprocess
from pathlib import Path

def git_status(repo_path: Path) -> dict:
    """Return a summary of the git status of one repo."""
    try:
        # --porcelain gives machine-readable output
        result = subprocess.run(
            ["git", "status", "--porcelain", "-b"],
            cwd=repo_path,
            capture_output=True,
            text=True,
            timeout=10,
            check=True
        )
    except subprocess.CalledProcessError as e:
        return {"name": repo_path.name, "error": e.stderr.strip()}
    except subprocess.TimeoutExpired:
        return {"name": repo_path.name, "error": "timed out"}
    except FileNotFoundError:
        return {"name": repo_path.name, "error": "git not installed"}

    lines = result.stdout.splitlines()
    branch = lines[0].replace("## ", "") if lines else "?"
    changes = [l for l in lines[1:] if l.strip()]

    return {
        "name":    repo_path.name,
        "branch":  branch,
        "dirty":   len(changes) > 0,
        "changes": len(changes)
    }

# Scan multiple repos
repos = [Path("work/api"), Path("work/web"),
         Path("work/docs"), Path("work/blog")]

for repo in repos:
    info = git_status(repo)
    if "error" in info:
        print(f"{info['name']:12s}  ✗ {info['error']}")
    else:
        status = "DIRTY" if info["dirty"] else "clean"
        print(f"{info['name']:12s}  {status:6s}  ({info['changes']} changes)  [{info['branch']}]")

def git_status(repo_path: Path) -> dict:
    """Return a summary of the git status of one repo."""
    if not repo_path.is_dir():
        return {"name": repo_path.name, "error": "folder not found"}
    try:
        ...                               # the rest is unchanged

Real-World Example — Live Log Tailing 📜असली Example — Live Log Tailing 📜真实案例 — 实时跟踪日志 📜

This tool shows the last lines of a log file and then follows new ones live — like tail -f — colouring each line by its level. We ran it on a sample app.log and pressed Ctrl+C after 1.5 seconds:

यह tool log file की आख़िरी lines दिखाता है और फिर नई lines को live follow करता है — tail -f की तरह — हर line को उसके level के रंग में। हमने इसे sample app.log पर चलाया और 1.5 seconds बाद Ctrl+C दबाया:

这个工具先显示日志文件的最后几行，然后像 tail -f 一样实时跟踪新行，并按级别给每行上色。我们在示例 app.log 上运行它，1.5 秒后按下 Ctrl+C：

import subprocess

def colorize(line: str) -> str:
    if "ERROR" in line:
        return f"\033[31m{line}\033[0m"       # red
    if "WARN" in line:
        return f"\033[33m{line}\033[0m"       # yellow
    if "INFO" in line:
        return f"\033[36m{line}\033[0m"       # cyan
    return line

def tail_log(path: str, lines: int = 100):
    p = subprocess.Popen(
        ["tail", "-n", str(lines), "-f", path],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )
    try:
        for line in p.stdout:
            print(colorize(line.rstrip()))
    except KeyboardInterrupt:
        print("\nStopping tail…")
        p.terminate()                        # polite SIGTERM first
        try:
            p.wait(timeout=2)
        except subprocess.TimeoutExpired:
            p.kill()                         # SIGKILL if it ignores us
            p.wait()
        print("tail exit code:", p.returncode)

tail_log("app.log")

🛑 Simulator: Shutdown Lab🛑 Simulator: Shutdown Lab🛑 模拟器：关闭实验室

Pick a child — one that cleans up on SIGTERM, one with no handler, and one that ignores SIGTERM — and a way to stop it. Watch the signals, the waiting and the final returncode. All six combinations were run for real.

एक child चुनिए — एक जो SIGTERM पर सफ़ाई करता है, एक जिसमें कोई handler नहीं, और एक जो SIGTERM अनसुना करता है — और उसे रोकने का तरीक़ा। Signals, इंतज़ार और अंतिम returncode देखिए। सभी छह combinations असल में चलाए गए।

选一个子进程 — 收到 SIGTERM 会清理的、没有处理函数的、忽略 SIGTERM 的 — 再选一种停止方式。观察信号、等待过程和最终的 returncode。全部六种组合都真实运行过。

run vs Popen — Which to Use When ⚖️run बनाम Popen — कब कौन-सा ⚖️run vs Popen — 何时用哪个 ⚖️

Common Pitfalls (and Good Habits) 🩹आम ग़लतियाँ (और अच्छी आदतें) 🩹常见陷阱（与好习惯）🩹

🩺 Simulator: Bug Doctor🩺 Simulator: Bug Doctor🩺 模拟器：Bug 医生

Six real subprocess mistakes. Buggy code on the left, fixed code on the right, changed lines marked. Every output is from a real run.

Subprocess की छह असली ग़लतियाँ। बाईं ओर buggy code, दाईं ओर ठीक किया हुआ, बदली lines चिह्नित। हर output असली run से है।

六个真实的 subprocess 错误。左边是有 bug 的代码，右边是修正后的代码，改动的行已标出。每个输出都来自真实运行。

Golden Rules and Quiz 🏆सुनहरे नियम और Quiz 🏆黄金法则与小测验 🏆

Quick Quiz — 8 Questionsछोटा Quiz — 8 सवाल小测验 — 8 道题