本文介绍如何在http请求成功或失败后，依据预定义的输出标识（如"black"/"white"）自动调用对应外部python脚本（如test1.py或test2.py），核心采用subprocess.run实现安全、可控的子进程调度。
本文介绍如何在http请求成功或失败后，依据预定义的输出标识（如"black"/"white"）自动调用对应外部python脚本（如test1.py或test2.py），核心采用subprocess.run实现安全、可控的子进程调度。
在实际自动化流程中，常需根据上游操作的结果（如API响应状态）触发下游不同任务。例如：API调用成功时执行数据同步脚本（test1.py），失败时启动告警与回滚脚本（test2.py）。直接依赖try/except的返回值判断存在局限性（如return会跳过后续逻辑），更健壮的做法是解耦状态判定与脚本调度，并统一在确定输出后执行。
✅ 推荐方案：基于HTTP状态码的显式分支 + subprocess.run
避免过度依赖异常捕获（RequestException覆盖范围广，可能掩盖预期非2xx响应），推荐优先检查HTTP响应状态码，逻辑清晰且易于测试：
import subprocess
import requests
import logging
logger = logging.getLogger(__name__)
def run_script_by_api_result(url, group, headers, auth):
# 定义输出标识到脚本路径的映射
script_map = {
"BLACK": "./test1.py", # 成功路径
"WHITE": "./test2.py" # 失败路径
}
try:
response = requests.post(
url, json=group, headers=headers, auth=auth, timeout=30
)
if response.status_code in (200, 201): # 可按需扩展允许的状态码
logger.info(f"API Response: {response.text}")
output_flag = "BLACK"
else:
logger.error(
f"Failed to create group '{group.get('groupName', 'unknown')}'. "
f"HTTP {response.status_code}: {response.reason}"
)
output_flag = "WHITE"
except requests.exceptions.Timeout:
logger.error("Request timed out")
output_flag = "WHITE"
except requests.exceptions.ConnectionError:
logger.error("Connection failed")
output_flag = "WHITE"
except requests.exceptions.RequestException as e:
logger.error(f"Unexpected request error: {e}")
output_flag = "WHITE"
# 统一执行对应脚本
script_path = script_map[output_flag]
try:
result = subprocess.run(
["python", script_path], # 推荐使用列表形式，避免shell注入
capture_output=True,
text=True,
timeout=60 # 防止子脚本无限阻塞
)
logger.info(f"Executed {script_path}, return code: {result.returncode}")
if result.stdout:
logger.debug(f"Script stdout: {result.stdout}")
if result.stderr:
logger.warning(f"Script stderr: {result.stderr}")
except subprocess.TimeoutExpired:
logger.error(f"Script {script_path} timed out")
except FileNotFoundError:
logger.error(f"Script not found: {script_path}. Check path and Python executable.")
except Exception as e:
logger.error(f"Failed to run {script_path}: {e}")
return output_flag == "BLACK" # 返回布尔值表示主流程是否成功⚠️ 关键注意事项
-
安全性：始终使用
subprocess.run([...])的列表形式传参，禁用shell=True，防止命令注入； -
可移植性：
["python", ...]依赖系统PATH；生产环境建议用sys.executable替代"python"，确保与当前解释器一致：import sys cmd = [sys.executable, script_path]
-
错误处理：为
subprocess.run添加timeout和异常捕获，避免主程序被卡死； -
日志透明化：通过
capture_output=True, text=True获取子脚本输出，便于调试； -
状态码设计：
response.status_code in (200, 201)比== 200更符合REST实践，可根据API文档灵活调整。
✅ 替代方案：finally 块（仅适用于简单场景）
若必须严格区分try（成功）与except（失败）路径，且不涉及复杂异常分类，可使用finally确保脚本总被执行：
def run_with_finally(url, group, headers, auth):
output_flag = "WHITE" # 默认失败态
try:
response = requests.post(url, json=group, headers=headers, auth=auth)
if response.status_code in (200, 201):
logger.info(f"API Response: {response.text}")
output_flag = "BLACK"
else:
raise requests.exceptions.HTTPError(f"Bad status: {response.status_code}")
except requests.exceptions.RequestException:
pass # 保持 output_flag = "WHITE"
finally:
script = ["python", "./test1.py" if output_flag == "BLACK" else "./test2.py"]
subprocess.run(script)总结：优先采用显式状态码分支而非宽泛异常捕获，结合
subprocess.run的安全参数配置，可构建高可靠性、易维护的脚本调度逻辑。务必对子进程执行添加超时与错误兜底，保障主流程稳定性。立即学习“Python免费学习笔记（深入）”；
