A Quick Blog Post: Orphaned Deployment Pipelines

You are not alone!

I recently completed some work for a client regarding orphaned deployment pipelines. In small teams, where one or two developers handle most of the tasks, certain items may become orphaned, and accessing them can be challenging.

While using REST API to achieve many things is common, working with PowerShell has never been my strong suit. Fortunately, this can be accomplished quite easily with Fabric and Fabric Notebooks.

Below is a quick code snippet you can run using a Fabric Python Notebook. Please note that users must have the appropriate permissions to read and make changes, such as being a Fabric Admin or having `Pipeline. ReadWrite. All` access.

REST API used in the example - Https://learn.microsoft.com/en-us/rest/api/power-bi/pipelines/update-pipeline-user

The Solution

Import sempy.fabric as fabric
import pandas as pd

# Initialize the Power BI REST API client
PBI_Rest_API = fabric. PowerBIRestClient()

try:
 # Define the URL for fetching deployment pipelines
 url = f'/v1.0/myorg/admin/pipelines'
 
 # Fetch the deployment pipelines data
 deployment_pipelines_df = PBI_Rest_API.get(url).json()['value']
 
 # Normalize the JSON data and convert it to a DataFrame
 deployment_pipelines_df = pd.json_normalize(deployment_pipelines_df)
 deployment_pipelines_df = pd. DataFrame(deployment_pipelines_df)
except Exception as e:
 # Print any exceptions that occur
 print(e)

# Display the DataFrame
display(deployment_pipelines_df)

# Iterate over the DataFrame rows
for index, row in deployment_pipelines_df.iterrows():
 # Extract pipeline name and ID from the current row
 Pipeline_Name = row['displayName']
 pipelineId = row['id']
 
 # Print the pipeline name and ID
 print("-" * 50)
 print(f"Processing Deployment Pipeline: {Pipeline_Name} (ID: {pipelineId})")
 
 # Add user to the deployment pipeline
 try:
 # Define the URL for adding a user to the pipeline
 add_user_url = f'/v1.0/myorg/admin/pipelines/{pipelineId}/users'
 
 # Define the user data to be added
 user_data = {
 "identifier": "prathy@data-nova.io",
 "accessRight": "Admin",
 "principalType": "User"
 }
 
 # Send a POST request to add the user
 response = PBI_Rest_API.post(add_user_url, json=user_data)
 
 # Check the response status code
 if response.status_code == 200:
 print(f"User added to pipeline {Pipeline_Name} successfully.")
 else:
 print(f"Failed to add user to pipeline {Pipeline_Name}. Status code: {response.status_code}")
 except Exception as e:
 # Print any exceptions that occur while adding the user
 print(f"Error adding user to pipeline {Pipeline_Name}: {str(e)}")

Get it from GitHub - Https://github.com/prathyusha-kamasani/Blog-Posts/blob/f3fd8d261860d52bee9a3e1b5d40202bdf8d564e/NB_Add_Users_Orphaned_Pipelines

Keep Smiling,

Prathy