如何使用Python循环调用HTTP接口验证GitLab流水线状态？

你的思路完全没问题——用while循环轮询状态确实比固定等待更高效灵活，但有几个细节需要调整才能正常工作，我来帮你梳理下：

首先，你写的循环条件里用了||，这是其他语言（比如Java/JS）的逻辑或运算符，Python里对应的是or，这个语法错误会导致代码跑不起来。

更重要的是：你的循环里没有重新调用API获取最新状态！如果只在循环外请求一次，returned_status会一直是初始的"pending"或"running"，循环会无限执行下去。必须把API请求的逻辑放到循环内部，每次循环都去拉取最新的状态。

另外，为了避免程序无限挂着（比如流水线卡住了），最好加上超时控制；同时处理API请求可能出现的异常（比如网络波动、权限问题），让代码更健壮。

下面是调整后的完整代码，我加了详细的注释：

import requests
import json
import time

def wait_for_pipeline_finish(pipeline_base_url, pipeline_id, api_headers, max_wait_seconds=300):
    # 拼接完整的流水线API地址
    pipeline_url = f"{pipeline_base_url}{pipeline_id}"
    start_time = time.time()

    while True:
        # 检查是否超过最大等待时间
        elapsed_time = time.time() - start_time
        if elapsed_time > max_wait_seconds:
            raise TimeoutError(f"等待流水线完成超时，已等待{elapsed_time:.1f}秒（上限{max_wait_seconds}秒）")
        
        try:
            # 每次循环都重新请求API，获取最新状态
            response = requests.get(url=pipeline_url, verify=False, headers=api_headers)
            # 如果请求返回HTTP错误（比如401、404、500），直接抛出异常
            response.raise_for_status()
            # 直接用response.json()解析返回结果，比手动解码更简洁
            pipeline_data = response.json()
            current_status = pipeline_data["status"]
            
            print(f"流水线ID {pipeline_id} 当前状态: {current_status}")

            # 判断是否进入最终状态
            if current_status in ["success", "failed"]:
                print(f"流水线完成！最终状态: {current_status}")
                return current_status
            
            # 未完成则等待5秒后重试
            time.sleep(5)
        
        # 捕获请求相关的异常（网络问题、HTTP错误等）
        except requests.exceptions.RequestException as e:
            print(f"请求API失败: {str(e)}，5秒后重试...")
            time.sleep(5)

# 调用示例（替换成你自己的变量）
triggered_pipeline_id = "7532"
pipeline_base_url = "https://your-gitlab-instance/api/v4/projects/xxx/pipelines/"  # 替换为你的GitLab API前缀
api_headers = {"PRIVATE-TOKEN": "your-access-token"}  # 替换为你的API凭证

try:
    final_status = wait_for_pipeline_finish(pipeline_base_url, triggered_pipeline_id, api_headers)
except TimeoutError as te:
    print(te)

几个关键优化点： #

• 状态实时更新：把API请求放到循环内部，每次都拉取最新状态
• 语法修正：用Python的or替代||，同时用in来判断状态是否在目标列表里，代码更简洁
• 超时控制：设置max_wait_seconds（默认5分钟），避免无限等待
• 异常处理：捕获网络请求的各种异常，出错后自动重试
• 代码可读性：用f-string拼接URL，用函数封装逻辑，更易维护

你原来的代码思路是对的，只要补上这些细节就能正常工作啦！

内容的提问来源于stack exchange，提问作者Štefan Sedmák