subprocess — Running External Programs

import subprocess

# Send input to a program's stdin
result = subprocess.run(
    ["grep", "error"],
    input="line 1\nerror line\nline 3\n",   # fed to stdin
    capture_output=True,
    text=True,
)
print(result.stdout)   # error line

# Set a timeout — kill the process if it runs too long
try:
    subprocess.run(["sleep", "10"], timeout=2)
except subprocess.TimeoutExpired:
    print("Command timed out after 2 seconds")

# Set the working directory and environment
result = subprocess.run(
    ["pwd"],
    cwd="/tmp",                          # run in this directory
    capture_output=True, text=True,
)
print(result.stdout.strip())   # /tmp

import os
custom_env = {**os.environ, "MY_VAR": "value"}
subprocess.run(["printenv", "MY_VAR"], env=custom_env)

# Pipelines — connect one program's output to another's input
# Equivalent to: ls /usr/bin | grep python
ls = subprocess.run(["ls", "/usr/bin"], capture_output=True, text=True)
grep = subprocess.run(
    ["grep", "python"],
    input=ls.stdout,         # feed ls output into grep
    capture_output=True, text=True,
)
print(grep.stdout)

# For true streaming pipelines, use Popen
p1 = subprocess.Popen(["ls", "/usr/bin"], stdout=subprocess.PIPE)
p2 = subprocess.Popen(["grep", "python"], stdin=p1.stdout,
                      stdout=subprocess.PIPE, text=True)
p1.stdout.close()   # allow p1 to receive SIGPIPE if p2 exits
output, _ = p2.communicate()
print(output)

import subprocess

# DANGEROUS — command injection vulnerability
filename = input("Enter filename: ")   # user types: x.txt; rm -rf /
subprocess.run(f"cat {filename}", shell=True)   # NEVER DO THIS
# The shell interprets ; rm -rf / as a second command!

# SAFE — arguments as a list, no shell
filename = input("Enter filename: ")
subprocess.run(["cat", filename])   # filename is a single argument
# Even "x.txt; rm -rf /" is treated as one (invalid) filename — safe

# If you MUST use shell features (pipes, globbing, redirection in the
# shell), sanitise input with shlex.quote — but prefer the list form:
import shlex
safe_arg = shlex.quote(filename)

# The list form also handles spaces and special characters correctly:
subprocess.run(["cat", "file with spaces.txt"])   # works perfectly
# vs shell=True where you'd need careful quoting

import subprocess

# Popen — start a process and interact with it while it runs
proc = subprocess.Popen(
    ["ping", "-c", "5", "example.com"],
    stdout=subprocess.PIPE,
    text=True,
)

# Stream output line by line as the process produces it
for line in proc.stdout:
    print(f"Got: {line.strip()}")

proc.wait()   # wait for the process to finish
print(f"Exit code: {proc.returncode}")

# poll() — check if the process is done without blocking
import time
proc = subprocess.Popen(["sleep", "3"])
while proc.poll() is None:    # None means still running
    print("Still running...")
    time.sleep(1)
print("Done")

# Async subprocess with asyncio
import asyncio

async def run_command(cmd):
    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    return stdout.decode()

async def main():
    # Run multiple commands concurrently
    results = await asyncio.gather(
        run_command(["echo", "first"]),
        run_command(["echo", "second"]),
        run_command(["echo", "third"]),
    )
    print(results)

asyncio.run(main())