本文介绍如何在 Python 中根据 HTTP 请求成功（如返回 "BLACK"）或失败（如返回 "WHITE"）的状态，自动调用对应外部脚本（如 test1.py 或 test2.py），核心使用 subprocess.run + 逻辑分支控制流程。
本文介绍如何在 python 中根据 http 请求成功（如返回 "black"）或失败（如返回 "white"）的状态，自动调用对应外部脚本（如 test1.py 或 test2.py），核心使用 `subprocess.run` + 逻辑分支控制流程。
在实际自动化或运维场景中，常需依据上游服务调用结果触发下游不同处理逻辑——例如 API 创建群组成功时执行通知脚本（test1.py），失败时启动告警或回滚脚本（test2.py）。直接依赖 try/except 的异常流来决定执行路径虽可行，但存在可读性差、难以扩展、且无法覆盖非异常型失败（如 HTTP 400/500 响应）等问题。推荐采用显式状态判断 + subprocess.run 执行外部脚本的稳健方案。
✅ 推荐做法：基于响应状态码的清晰分支控制
以下代码将 requests.post 的结果解耦为明确的状态变量（output），再通过映射表驱动脚本执行，逻辑清晰、易于维护和测试：
import subprocess as sp
import requests
import logging
logger = logging.getLogger(__name__)
def run_script_by_api_result(url, group, headers, auth):
# 定义输出状态与脚本路径的映射关系（支持灵活扩展）
script_mapping = {
"BLACK": "./test1.py",
"WHITE": "./test2.py"
}
try:
response = requests.post(
url, json=group, headers=headers, auth=auth, timeout=10
)
# 根据 HTTP 状态码判断业务成功与否（比捕获异常更精准）
if response.status_code in (200, 201): # 可按需添加 204 等合法状态
logger.info(f"API Response: {response.text}")
output = "BLACK"
else:
logger.error(
f"Failed to create group '{group.get('groupName', 'N/A')}'. "
f"HTTP {response.status_code}: {response.reason}"
)
output = "WHITE"
except requests.exceptions.Timeout:
logger.error("Request timed out.")
output = "WHITE"
except requests.exceptions.ConnectionError:
logger.error("Connection failed.")
output = "WHITE"
except requests.exceptions.RequestException as e:
logger.error(f"Unexpected request error: {e}")
output = "WHITE"
# 统一执行对应脚本
script_path = script_mapping[output]
logger.info(f"Executing script: {script_path}")
try:
result = sp.run(["python", script_path], capture_output=True, text=True, timeout=30)
logger.info(f"Script exited with code {result.returncode}")
if result.stdout:
logger.debug(f"Script stdout: {result.stdout.strip()}")
if result.stderr:
logger.warning(f"Script stderr: {result.stderr.strip()}")
return output == "BLACK" # 成功返回 True，否则 False 或 None
except FileNotFoundError:
logger.error("Python interpreter not found. Ensure 'python' is in PATH.")
raise
except sp.TimeoutExpired:
logger.error("Script execution timed out.")
raise
except Exception as e:
logger.error(f"Failed to run script: {e}")
raise⚠️ 关键注意事项
-
避免滥用
finally：原建议中用finally执行脚本看似简洁，但会强制运行（即使try中已return），且无法感知except中是否已设置output变量，易引发UnboundLocalError。显式分支更安全。 -
subprocess.run参数建议：- 使用
["python", "script.py"]列表形式传参，避免 shell 注入风险； - 添加
timeout防止子进程卡死； - 使用
capture_output=True, text=True便于日志记录； - 检查
result.returncode判断脚本是否真正成功。
- 使用
-
状态判定优先级：HTTP 状态码（如
200/201）比异常捕获更能反映业务语义；requests.exceptions.RequestException应仅用于网络层故障（超时、DNS 失败等），而非业务错误。 -
路径与环境：确保
test1.py/test2.py路径正确，且目标脚本具有可执行权限（Linux/macOS）或由 Python 解释器正确加载（Windows）。
✅ 总结
不要让异常处理承担业务路由职责。通过将 API 结果抽象为明确的状态标识（如 "BLACK"/"WHITE"），再结合字典映射与 subprocess.run，即可实现高可读、易测试、强健可控的跨脚本调度逻辑。此模式也易于演进为配置驱动（如从 JSON/YAML 加载映射表）或集成进 Celery/Airflow 等工作流系统。
