在 Python 3.2 中如何实现类似 `subprocess.run()` 的功能?
在 Python 3.2 中,`subprocess.run()` 函数尚未引入,因此需要使用 `subprocess.Popen` 来实现类似功能。`Popen` 提供了更底层的接口,通过组合使用 `communicate()` 方法和适当的参数,可以模拟 `subprocess.run()` 的行为。
以下是一个兼容 Python 3.2 的替代实现:
```python
import subprocess
def run_command(cmd):
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout, stderr = process.communicate()
return {
'returncode': process.returncode,
'stdout': stdout.decode('utf-8'),
'stderr': stderr.decode('utf-8')
}
# 示例:执行 echo 命令并输出结果
result = run_command(['echo', 'Hello World'])
print("标准输出:", result['stdout'])
print("错误输出:", result['stderr'])
print("退出码:", result['returncode'])
```
上述代码中,`Popen` 用于启动子进程,`communicate()` 用于等待子进程完成并读取其输出。由于 Python 3.2 不支持 `text=True` 或 `capture_output=True`,因此必须手动调用 `.decode()` 对字节流进行解码。
### 超时控制的实现
如果需要对命令执行设置超时限制,可以通过轮询 `process.poll()` 的返回值来判断子进程是否仍在运行,并在超时后调用 `kill()` 终止进程,从而实现超时控制:
```python
import subprocess
import time
def run_command_with_timeout(cmd, timeout=60):
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
start_time = time.time()
while process.poll() is None:
if time.time() - start_time > timeout:
process.kill()
raise Exception("命令执行超时")
stdout, stderr = process.communicate()
return {
'returncode': process.returncode,
'stdout': stdout.decode('utf-8'),
'stderr': stderr.decode('utf-8')
}
# 示例:执行带有超时的命令
try:
result = run_command_with_timeout(['sleep', '5'], timeout=3)
except Exception as e:
print("发生异常:", e)
```
该实现通过 `time.time()` 记录开始时间,并在每次轮询时检查是否超过设定的超时时间,若超时则终止子进程并抛出异常。
###
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考