一文详解Python中的subprocess模块

目录

• subprocess模块简介
• 常用函数
• 执行外部命令
• 使用subprocess.run()
• 使用subprocess.call()
• 使用subprocess.check_call()
• 使用subprocess.check_output()
• 管道通信
• 子进程管理
• 终止子进程
• 获取子进程的返回码
• 错误处理
• 捕获异常
• 超时处理
• 实际应用示例
• 运行shell命令
• 备份数据库
• 自动化脚本
• 最佳实践
• 结论

import subprocess

# 运行一个命令并等待其完成
subprocess.run(["ls", "-l"])

# 使用Popen对象启动和管理进程
process = subprocess.Popen(["ls", "-l"], stdout=subprocess.PIPE)
output, error = process.communicate()
print(output.decode())

import subprocess

# 执行命令并等待完成
result = subprocess.run(["echo", "Hello, World!"], capture_output=True, text=True)
print(result.stdout)

import subprocess

# 执行命令并返回退出状态
return_code = subprocess.call(["ls", "-l"])
print(f"Return code: {return_code}")

import subprocess

try:
    subprocess.check_call(["ls", "-l"])
except subprocess.CalledProcessError as e:
    print(f"Command failed with return code {e.returncode}")

import subprocess

try:
    output = subprocess.check_output(["echo", "Hello, World!"], text=True)
    print(output)
except subprocess.CalledProcessError as e:
    print(f"Command failed with return code {e.returncode}")

import subprocess

# 将子进程的输出重定向到父进程
process = subprocess.Popen(["echo", "Hello, World!"], stdout=subprocess.PIPE)
output, error = process.communicate()
print(output.decode())

import subprocess

# 使用管道连接两个命令
p1 = subprocess.Popen(["echo", "Hello, World!"], stdout=subprocess.PIPE)
p2 = subprocess.Popen(["grep", "Hello"], stdin=p1.stdout, stdout=subprocess.PIPE)
p1.stdout.close()  # 允许p1关闭它的输出管道
output, error = p2.communicate()
print(output.decode())

import subprocess
import time

# 启动一个长时间运行的进程
process = subprocess.Popen(["sleep", "10"])

# 等待2秒后终止进程
time.sleep(2)
process.terminate()
process.wait()
print("Process terminated")

import subprocess

process = subprocess.Popen(["ls", "-l"], stdout=subprocess.PIPE)
process.wait()
print(f"Return code: {process.returncode}")

import subprocess

try:
    subprocess.check_call(["false"])
except subprocess.CalledProcessError as e:
    print(f"Command failed with return code {e.returncode}")

import subprocess

try:
    subprocess.run(["sleep", "10"], timeout=5)
except subprocess.TimeoutExpired:
    print("Command timed out")

import subprocess

result = subprocess.run("ls -l | grep .py", shell=True, capture_output=True, text=True)
print(result.stdout)

import subprocess

command = ["mysqldump", "-u", "root", "-p", "database_name"]
with open("backup.sql", "w") as f:
    result = subprocess.run(command, stdout=f)
    if result.returncode == 0:
        print("Backup successful")
    else:
        print("Backup failed")

import subprocess

def run_commands(commands):
    for command in commands:
        result = subprocess.run(command, shell=True, capture_output=True, text=True)
        if result.returncode != 0:
            print(f"Command failed: {command}")
            print(result.stderr)
            break
        else:
            print(result.stdout)

commands = [
    "echo 'Starting automation script'",
    "mkdir -p /tmp/test_directory",
    "touch /tmp/test_directory/test_file",
    "echo 'Automation script completed'"
]

run_commands(commands)

import subprocess

def safe_run_command(command, timeout=10):
    try:
        result = subprocess.run(command, capture_output=True, text=True, timeout=timeout)
        result.check_returncode()
        return result.stdout
    except subprocess.CalledProcessError as e:
        print(f"Command '{e.cmd}' failed with return code {e.returncode}")
    except subprocess.TimeoutExpired as e:
        print(f"Command '{e.cmd}' timed out after {e.timeout} seconds")

output = safe_run_command(["ls", "-l"])
print(output)