лазурный конвейер для удаления старой лазурной ветки git (не репо)

Я использую rest azure rest api для вызова ветки, но не могу удалить в соответствии с параметрами даты.

все работает, за исключением того, что пользовательский ввод не работает в лазурном конвейере, который я принял как жесткий код.

Для пользовательского ввода (вводимых учетных данных), пожалуйста, обратитесь к приведенному ниже образцу:

import requests
import base64

repo_endpoint_url = "https://dev.azure.com/<organization>/<project>/_apis/git/repositories?api-version=5.1"

username = "" # This can be an arbitrary value or you can just let it empty
password = "<your-password-here>"
userpass = username + ":" + password
b64 = base64.b64encode(userpass.encode()).decode()
headers = {"Authorization" : "Basic %s" % b64} 

response = requests.get(repo_endpoint_url, headers=headers)
print(response.status_code) # Expect 200

Вы также можете попробовать использовать токен PAT или OAuth $env:SYSTEM_ACCESSTOKEN напрямую (используйте PAT или $ env: SYSTEM_ACCESSTOKEN для замены password).

Однако, чтобы ваш скрипт мог использовать токен OAuth конвейера сборки, нам нужно перейти на вкладку Параметры конвейера сборки и выбрать Allow Scripts to Access OAuth Token.

Я использую rest azure rest api для вызова ветки, но не могу удалить в соответствии с параметрами даты.

import requests
import sys
from datetime import datetime as dt
import json
from git import Repo
import git
import time

username = '<u name>'
auth_key = '<auth key>'

class gitRepoDeletion:

    def getRepo(self, organization_name, project_name, repo_name):
        """
        Getting the repo details
        from the user and
        flitering the master
        branch with date functionality(still implementing)
        """
        getting_repo_list = "https://dev.azure.com/" + organization_name + '/' + \
            project_name + "/_apis/git/repositories/" + repo_name + "/refs?api-version=5.0"

        get_reponse = requests.get(
            getting_repo_list, auth=(username,auth_key))

        try:
            repojson = json.loads(get_reponse.content)
        except ValueError:
            print("Error loading json file")
        output_json = [x for x in repojson['value']
                       if x['name'] != 'refs/heads/master']
        with open('/home/vsts/work/1/s/data.json', 'w', encoding='utf-8') as f:
            json.dump(output_json, f, ensure_ascii=False, indent=4)

    def filtering_branches(self, organization_name, project_name, repo_name, user_date):
        """ 
        Filtering branches according 
        to the date passed by user

        """

        git_url = "https://" + organization_name + "@dev.azure.com" + '/' + \
            organization_name + '/' + project_name + '/_git' + '/' + repo_name

        branches = Repo.clone_from(git_url, "./mylocaldir209")

        remote_branches = []
        for ref in branches.git.branch('-r').split('\n'):
            if ref != '  origin/HEAD -> origin/master':
                if ref != '  origin/master':
                    remote_branches.append(ref[9:])
                else:
                    pass

        branch_and_timing_dict = {}

        for listy in remote_branches:
            branches.git.checkout(listy)
            commit = branches.head.commit
            timing = time.asctime(time.gmtime(commit.committed_date))
            timing = time.strftime(
                "%d/%m/%Y", time.gmtime(commit.committed_date)).replace(' 0', ' ')
            branch_and_timing_dict[listy] = timing

        global filterlist
        filterlist = []

        for key, values in branch_and_timing_dict.items():
            d1 = dt.strptime(user_date, "%d/%m/%Y") 
            d2 = dt.strptime(key, "%d/%m/%Y")
            if d1 > d2:
                filterlist.append(values)
            else:
                pass
        return filterlist

    def repo_delete(self, organization_name, project_name, repo_name, dry_flag):
        """
        Deleting repo as
        per date input by user
        also exlucling master

        """

        all_repo_to_be_deleted = []

        newObjectId = "0000000000000000000000000000000000000000"

        filteredBranchesAsPerDateWithRef = []

        for value in filterlist:
            filteredBranchesAsPerDateWithRef.append("refs/heads/" + value)
            print(value)
        print(filteredBranchesAsPerDateWithRef)

        # Cluttering extra spaces and lowering the case of the dry flag value passed by the user

        # Reading data.json file, which is fetched by the getRepo() method after excluding the master branch
        with open('/home/vsts/work/1/s/data.json') as data_file:
            json_data = json.load(data_file)
        for item in json_data:
            name_of_branch = item['name']
            objectId = item['objectId']
            # Adding name_of_branch in all_repo_to_be_deleted list
            all_repo_to_be_deleted.append(name_of_branch)
            # Adding objectId in all_repo_to_be_deleted list
            # all_repo_to_be_deleted.append(objectId)
            passing_branch_name = "https://dev.azure.com/" + organization_name + '/' + \
                project_name + "/_apis/git/repositories/" + repo_name + "/refs?api-version=5.0"
            headers = {'Content-type': 'application/json'}

        for nameOfBranchWithref in filteredBranchesAsPerDateWithRef:
            print(nameOfBranchWithref)
            nameOfBranchWithref = nameOfBranchWithref

            data = [
                {
                    "name": nameOfBranchWithref,
                    "newObjectId": newObjectId,
                    "oldObjectId": objectId,
                }
            ]

            dry_flag = dry_flag.lower().strip()
            if dry_flag == 'true':
                repo_delete = requests.post(passing_branch_name, data=json.dumps(
                    data), headers=headers, auth=(username, auth_key))
                print(repo_delete)

            else:
                with open('delete_repo.txt', 'w') as d:
                    for item in all_repo_to_be_deleted:
                        d.write("%s\n" % item)
                print("---- This is Dry Run ----")
                print("These are the repo to be deleted: ", all_repo_to_be_deleted)


if __name__ == "__main__":
    gitRepoDeletion().getRepo('sushmasureshyadav202', 'my_delete_git', 'my_delete_git')

    gitRepoDeletion().filtering_branches(
        "<azure org name>", '<azure project>', '<azure repo>', "31/1/2020")
    gitRepoDeletion().repo_delete("<azure org name>", '<azure project>', '<azure repo>', 'true')