
本文介绍如何在 requests.post 成功时执行 test1.py,失败时执行 test2.py,通过状态判断 + subprocess.run 实现脚本的条件化调用,避免 try/except 过度耦合逻辑。
本文介绍如何在 `requests.post` 成功时执行 `test1.py`,失败时执行 `test2.py`,通过状态判断 + `subprocess.run` 实现脚本的条件化调用,避免 `try/except` 过度耦合逻辑。
在实际自动化流程中,常需根据上游 API 调用结果(如 HTTP 状态码或异常类型)触发下游不同行为。直接在 try/except 中嵌套脚本执行虽可行,但可读性差、难以维护,且 return 会跳过 except 后续逻辑。更健壮的做法是:将“结果判定”与“脚本执行”解耦,统一在确定输出状态后再调用目标脚本。
推荐采用基于 HTTP 状态码的显式分支逻辑(而非依赖异常捕获),因其语义清晰、可控性强、易于扩展。以下是完整实现:
import subprocess
import requests
import logging
logger = logging.getLogger(__name__)
def run_script_based_on_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
)
# 显式检查 HTTP 状态码(推荐:比异常捕获更精准)
if response.status_code in (200, 201): # 成功创建通常返回 200 或 201
logger.info(f"API Response: {response.text}")
output_state = "BLACK"
else:
logger.error(
f"Failed to create group '{group.get('groupName', 'unknown')}'. "
f"HTTP {response.status_code}: {response.reason}"
)
output_state = "WHITE"
except requests.exceptions.Timeout:
logger.error("Request timed out.")
output_state = "WHITE"
except requests.exceptions.ConnectionError:
logger.error("Connection failed.")
output_state = "WHITE"
except requests.exceptions.RequestException as e:
logger.error(f"Unexpected request error: {e}")
output_state = "WHITE"
# 统一执行对应脚本(无论成功/失败路径均走到此处)
script_path = script_map[output_state]
logger.info(f"Executing {script_path} based on state: {output_state}")
try:
result = subprocess.run(
["python", script_path],
capture_output=True,
text=True,
check=False # 不抛异常,便于处理脚本自身的退出码
)
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()}")
except FileNotFoundError:
logger.error(f"Python interpreter or script '{script_path}' not found.")
except Exception as e:
logger.error(f"Failed to launch script: {e}")
return output_state == "BLACK" # 返回 True 表示 API 成功(按原始语义)
✅ 关键优势说明:
-
解耦清晰:API 结果判定(
output_state)与脚本执行完全分离,逻辑主线单一; -
状态可扩展:只需修改
script_map字典即可新增状态(如"GRAY"→test3.py); -
错误更精准:区分
Timeout、ConnectionError等具体异常,便于针对性重试或告警; -
脚本执行健壮:使用
capture_output=True捕获日志,check=False避免因子脚本非零退出码中断主流程; -
安全可靠:显式传入
["python", script_path]列表形式调用,规避 shell 注入风险。
⚠️ 注意事项:
- 确保
test1.py和test2.py具有可执行权限(Linux/macOS 下运行chmod +x test1.py),或确保python命令在系统 PATH 中; - 若目标脚本需接收参数,可在
subprocess.run()的命令列表中追加,例如["python", script_path, "--env", "prod"]; - 生产环境建议为子进程设置
timeout参数(如timeout=60),防止脚本长期阻塞主流程; - 日志中应避免直接打印敏感字段(如
group全量内容),建议脱敏后记录关键标识(如groupName)。
通过该模式,你不仅能实现 BLACK/WHITE 的二元分发,更能轻松演进为多状态工作流引擎的基础组件。
大量免费API接口:立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游,为数据API应用能力赋能!











