触发 python azure 函数从密钥库获取秘密

Question

我尝试运行由 blob 触发的 python azure 函数，该函数将获取保存在密钥库中的个人访问令牌并运行 DevOps 管道。我在本地测试了代码，它工作正常，当我将代码包含在 init.py 文件中时，它不会触发管道。由于提供的信息不多，甚至无法调试代码。

以下是部署前在 init.py 文件中编写的代码，我在 requirements.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)

请指导我

Answer 1

在天蓝色函数的身份下创建系统分配/用户分配的访问对象ID（SP）。

创建具有对密钥库中上述 SP 的必要访问权限的访问策略