Azure DevOps主流水线：检查依赖流水线制品并按需触发运行

核心思路 #

通过Python脚本调用Azure DevOps REST API完成两个核心操作：

• 检查目标流水线对应分支的最新成功运行是否存在指定制品
• 若制品不存在，触发目标流水线的分支构建

步骤1：编写Python检查与触发脚本 #

脚本依赖requests库，逻辑如下：

import os
import requests
from requests.auth import HTTPBasicAuth

# 从流水线环境变量读取配置
ORGANIZATION = os.getenv("AZURE_DEVOPS_ORG")
PROJECT = os.getenv("TARGET_PROJECT")
PIPELINE_ID = os.getenv("TARGET_PIPELINE_ID")
BRANCH_NAME = os.getenv("BUILD_SOURCE_BRANCH")
ARTIFACT_NAME = "artifact-abc"
PAT_TOKEN = os.getenv("AZURE_DEVOPS_PAT")

BASE_API_URL = f"https://dev.azure.com/{ORGANIZATION}/{PROJECT}/_apis"

def check_artifact_exists():
    # 获取目标分支的最新成功构建记录
    runs_api = f"{BASE_API_URL}/build/builds?definitions={PIPELINE_ID}&branchName={BRANCH_NAME}&statusFilter=completed&resultFilter=succeeded&$top=1&api-version=7.1-preview.7"
    resp = requests.get(runs_api, auth=HTTPBasicAuth("", PAT_TOKEN))
    resp.raise_for_status()
    runs = resp.json()["value"]

    if not runs:
        print("目标分支无成功构建记录，制品不存在")
        return False
    
    latest_run_id = runs[0]["id"]
    # 检查该构建是否包含指定制品
    artifacts_api = f"{BASE_API_URL}/build/builds/{latest_run_id}/artifacts?api-version=7.1-preview.5"
    artifacts_resp = requests.get(artifacts_api, auth=HTTPBasicAuth("", PAT_TOKEN))
    artifacts_resp.raise_for_status()
    artifacts = artifacts_resp.json()["value"]

    for artifact in artifacts:
        if artifact["name"] == ARTIFACT_NAME:
            print(f"找到目标制品：{ARTIFACT_NAME}，来自构建ID {latest_run_id}")
            return True
    
    print(f"最新成功构建（ID {latest_run_id}）中无指定制品：{ARTIFACT_NAME}")
    return False

def trigger_pipeline():
    # 触发目标流水线的分支构建
    trigger_api = f"{BASE_API_URL}/pipelines/{PIPELINE_ID}/runs?api-version=7.1-preview.1"
    payload = {
        "resources": {
            "repositories": {
                "self": {
                    "refName": BRANCH_NAME
                }
            }
        }
    }
    resp = requests.post(trigger_api, json=payload, auth=HTTPBasicAuth("", PAT_TOKEN))
    resp.raise_for_status()
    run_info = resp.json()
    print(f"已触发目标流水线运行，运行ID：{run_info['id']}，状态：{run_info['state']}")

if __name__ == "__main__":
    artifact_exists = check_artifact_exists()
    if not artifact_exists:
        print("开始触发目标流水线构建...")
        trigger_pipeline()
        # 可选：若需等待构建完成再执行后续步骤，可在此添加轮询构建状态的逻辑

步骤2：在主流水线YAML中集成脚本 #

修改原有流水线YAML，添加脚本运行步骤，注意安全存储PAT令牌：

# 配置Python环境
- task: UsePythonVersion@0
  inputs:
    versionSpec: '3.x'
    addToPath: true

# 安装依赖库
- script: pip install requests
  displayName: 'Install required packages'

# 运行检查与触发脚本
- task: PythonScript@0
  inputs:
    scriptSource: 'filePath'
    scriptPath: '$(Build.SourcesDirectory)/check_and_trigger.py'  # 脚本存放路径
  env:
    AZURE_DEVOPS_ORG: '你的Azure DevOps组织名'
    TARGET_PROJECT: 'ABC'
    TARGET_PIPELINE_ID: '1'
    AZURE_DEVOPS_PAT: $(AzureDevOpsPAT)  # 从流水线秘密变量中读取PAT
  displayName: 'Check artifact existence and trigger pipeline'

# 原有的制品下载任务
- task: DownloadPipelineArtifact@2
  inputs:
    source: specific
    project: ABC
    pipeline: 1
    runVersion: latestFromBranch
    runBranch: $(Build.SourceBranch)
    artifact: artifact-abc
    targetPath: $(Pipeline.Workspace)

关键注意事项 #

• PAT权限：需为PAT分配Build (Read & Execute)权限，确保能读取构建记录和触发流水线
• 分支格式：BUILD_SOURCE_BRANCH默认格式为refs/heads/xxx，可直接传入API无需转换
• 构建等待（可选）：若需等待触发的构建完成再下载制品，可在脚本中添加轮询逻辑，检查构建状态直到变为成功
• API版本：示例使用较新的API版本，若遇兼容性问题可调整为对应版本

内容的提问来源于stack exchange，提问作者UnKnown777