如何根据API调用结果动态执行不同Python脚本-Python教程-PHP中文网

如何根据API调用结果动态执行不同Python脚本

本文介绍如何在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确保脚本总被执行：

python-parallelization

将顺序 Python 代码转换为并行/并发实现，用于在需要并行化 Python 代码、提升性能时使用。

下载

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)