向多个Azure Pipelines添加任务遇405错误，求自动化方案

问题描述 #

我尝试编写Python脚本向多个Azure Pipelines添加任务，已使用拥有全权限的Personal Access Token（PAT），但仍收到状态码405错误（提示请求资源不支持PUT方法）。需要修复该问题，并了解其他可实现批量添加任务的自动化方式。

我的代码 #

import requests
import base64

# Azure DevOps organization URL
org_url = "https://dev.azure.com/my-org"  # Replace with your organization URL

# Personal access token (PAT) with necessary permissions (Read and Write) for pipelines
pat = "my-pat" 

# Encode the PAT to Base64
credentials = base64.b64encode(f":{pat}".encode()).decode()

# Project name and pipeline ID where you want to add the tasks
project_name = "My Project"
pipeline_id = 12

# Tasks to be added
tasks = [
    {
        "taskReference": {
            "id": "mspremier.PostBuildCleanup.PostBuildCleanup-task.PostBuildCleanup@3",
            "name": "Clean Agent Directories"
        },
        "enabled": True
    },
    {
        "taskReference": {
            "id": "NodeTool@0",
            "name": "Use Node 6.x"
        },
        "enabled": True
    }
]

def update_pipeline_definition():
    url = f"{org_url}/{project_name}/_apis/pipelines/{pipeline_id}?api-version=6.0-preview.1"
    headers = {"Authorization": f"Basic {credentials}", "Content-Type": "application/json"}

    # Get the current pipeline definition
    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        pipeline_data = response.json()

        # Modify the pipeline definition to include the tasks
        if "phases" in pipeline_data["configuration"]:
            for phase in pipeline_data["configuration"]["phases"]:
                if "steps" in phase:
                    for task in tasks:
                        phase["steps"].append(task)

        # Update the pipeline definition
        response = requests.put(url, headers=headers, json=pipeline_data)

        if response.status_code == 200:
            print(f"Tasks added to pipeline {pipeline_id} successfully.")
        else:
            print(f"Failed to update pipeline {pipeline_id}. Status code: {response.status_code}")
            print(response.text)
    else:
        print(f"Failed to get pipeline {pipeline_id}. Status code: {response.status_code}")
        print(response.text)

if __name__ == "__main__":
    update_pipeline_definition()

报错信息 #

Failed to update pipeline 42. Status code: 405
{"count":1,"value":{"Message":"The requested resource does not support http method 'PUT'."}}

问题原因及修复方案 #

原因 #

你调用的/_apis/pipelines/{pipeline_id}端点仅用于获取流水线的元数据（如名称、ID、状态等），不支持PUT请求更新流水线定义。对于经典流水线（Classic Pipeline），正确的API端点是/_apis/build/definitions/{definitionId}。

修复后的代码 #

import requests
import base64

org_url = "https://dev.azure.com/my-org"
pat = "my-pat"
credentials = base64.b64encode(f":{pat}".encode()).decode()
project_name = "My Project"
pipeline_id = 12

tasks = [
    {
        "taskReference": {
            "id": "mspremier.PostBuildCleanup.PostBuildCleanup-task.PostBuildCleanup@3",
            "name": "Clean Agent Directories"
        },
        "enabled": True
    },
    {
        "taskReference": {
            "id": "NodeTool@0",
            "name": "Use Node 6.x"
        },
        "enabled": True
    }
]

def update_pipeline_definition():
    # 替换为经典流水线的API端点，使用稳定版API 6.0
    url = f"{org_url}/{project_name}/_apis/build/definitions/{pipeline_id}?api-version=6.0"
    headers = {"Authorization": f"Basic {credentials}", "Content-Type": "application/json"}

    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        pipeline_data = response.json()

        # 经典流水线的步骤位于process/phases/steps路径下
        if "process" in pipeline_data and "phases" in pipeline_data["process"]:
            for phase in pipeline_data["process"]["phases"]:
                if "steps" in phase:
                    phase["steps"].extend(tasks)

        # 使用PUT更新构建定义
        response = requests.put(url, headers=headers, json=pipeline_data)
        if response.status_code == 200:
            print(f"Tasks added to pipeline {pipeline_id} successfully.")
        else:
            print(f"Failed to update pipeline {pipeline_id}. Status code: {response.status_code}")
            print(response.text)
    else:
        print(f"Failed to get pipeline {pipeline_id}. Status code: {response.status_code}")
        print(response.text)

if __name__ == "__main__":
    update_pipeline_definition()

其他批量添加任务的自动化方式 #

1. 使用Azure DevOps CLI #

通过az pipelines build definition update命令可以直接修改流水线定义，结合脚本实现批量处理：

• 安装Azure CLI并登录：az login，然后配置DevOps组织：az devops configure --defaults organization=https://dev.azure.com/my-org project="My Project"
• 导出流水线定义到JSON文件：az pipelines build definition show --id {pipeline_id} > definition.json
• 修改JSON文件中的步骤后，更新定义：az pipelines build definition update --id {pipeline_id} --definition-file definition.json
• 批量处理可通过Shell/PowerShell脚本遍历流水线ID列表，重复上述步骤。

2. 使用Azure DevOps Python SDK #

官方SDK封装了API调用，更易维护：

• 安装SDK：pip install azure-devops
• 示例代码：

from azure.devops.connection import Connection
from msrest.authentication import BasicAuthentication

org_url = "https://dev.azure.com/my-org"
pat = "my-pat"
credentials = BasicAuthentication("", pat)
connection = Connection(base_url=org_url, creds=credentials)

build_client = connection.clients.get_build_client()
project_name = "My Project"
pipeline_id = 12

# 获取构建定义
definition = build_client.get_definition(project_name, pipeline_id)

# 添加任务到第一个阶段的步骤
tasks = [
    {
        "taskReference": {
            "id": "mspremier.PostBuildCleanup.PostBuildCleanup-task.PostBuildCleanup@3",
            "name": "Clean Agent Directories"
        },
        "enabled": True
    }
]
definition.process.phases[0].steps.extend(tasks)

# 更新定义
updated_definition = build_client.update_definition(definition, project_name)
print(f"Tasks added to pipeline {pipeline_id} successfully.")

3. YAML流水线模板化（适用于YAML流水线） #

如果使用YAML流水线，推荐将通用任务封装为模板：

• 创建模板文件tasks-template.yml：

steps:
- task: mspremier.PostBuildCleanup.PostBuildCleanup-task.PostBuildCleanup@3
  displayName: 'Clean Agent Directories'
- task: NodeTool@0
  displayName: 'Use Node 6.x'
  inputs:
    versionSpec: 6.x

• 在各个流水线的YAML文件中引用模板：

jobs:
- job: Build
  steps:
  - template: tasks-template.yml

• 通过Git批量修改所有流水线YAML文件的引用，提交后自动更新流水线定义。

内容的提问来源于stack exchange，提问作者P A G