Блог

Я попытался запустить функцию python azure, запускаемую blob-объектом, которая получит токен личного доступа, сохраненный в хранилище ключей, и запустит конвейер DevOps. Я протестировал код локально, и он работает нормально, и когда я включаю код в init.py файл он не запускает конвейер. Даже невозможно отладить код, поскольку предоставлено не так много информации.

Ниже приведен код, написанный в init.py файл перед развертыванием, и я предоставил необходимые библиотеки в requirement.txt файл

     import logging
    from azure.devops.connection import Connection
    from msrest.authentication import BasicAuthentication
    import azure.functions as func

    from azure.identity import ManagedIdentityCredential
    from azure.keyvault.secrets import SecretClient

    credentials = ManagedIdentityCredential()

    secret_client = SecretClient(vault_url="https://myKeyvault.vault.azure.net", credential=credentials)
    Personal_Access_Token = secret_client.get_secret("devops-token")
    print(Personal_Access_Token.value)

    Organization_URL = 'https://dev.azure.com/org/'
    Project_Name = 'ProjectName'

    def create_pipeline_client():
        credentials = BasicAuthentication('',Personal_Access_Token.value)
        connection = Connection(base_url=Organization_URL,creds=credentials)
        pipeline_client = connection.clients_v6_0.get_pipelines_client()
        return pipeline_client
        
    def build_pipeline(pipeline_id,run_params,pipeline_version=None):
        pipeline_client = create_pipeline_client()
        print("Running Pipeline with ID : "  str(pipeline_id))
        try:
            pipeline_client.run_pipeline(run_parameters=run_params,project=Project_Name,pipeline_id=pipeline_id,pipeline_version=pipeline_version)
            print("Pipeline Run sucessfully activated")
        except Exception as ex:
            print("Pipeline Failed with Exception : "   str(ex))


    def get_pipeline(pipeline_id,pipeline_version=None):
        pipeline_client = create_pipeline_client()
        pipeline = pipeline_client.get_pipeline(project=Project_Name,pipeline_id=pipeline_id,pipeline_version=pipeline_version)
        print(pipeline)
        
    def list_pipelines():
        pipeline_client = create_pipeline_client()
        pipeline_list = pipeline_client.list_pipelines(Project_Name)
        for item in pipeline_list:
            print(item)

    def main(myblob: func.InputStream):
        logging.info(f"Python blob trigger function processed blob n"
                     f"Name: {myblob.name}n"
                     f"Blob Size: {myblob.length} bytes")

        run_params = {'branch/tag':'master'}
        build_pipeline(1,run_params,None)
  

пожалуйста, направьте меня

создайте назначенный системе / назначенный пользователю идентификатор объекта доступа (SP) в разделе identity в функции azure.

Создайте политику доступа с необходимым доступом к вышеуказанному SP в хранилище ключей