I tried to run the python azure function triggered by the blob, which will get the Personal Access token saved in the key vault and run the DevOps pipeline. I tested the code locally and it works fine and when I include the code in the init.py file it is not triggering the pipeline. Even unable to debug the code as there is not much information provided.
Below is the code is written in the init.py file before deploying and i have provided the required libraries in the requirement.txt file
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)
please guide me
create system assigned/user assigned access object ID (SP) under identity in azure function.
Create Access policy with necessary access to the above SP in the key vault