deployment_pipelines

src/sempy_labs/__init__.py

@@ -439,9 +439,6 @@
     "resolve_capacity_name",
     "run_model_bpa_bulk",
     "create_model_bpa_semantic_model",
-    "list_deployment_pipeline_stage_items",
-    "list_deployment_pipeline_stages",
-    "list_deployment_pipelines",
     "get_git_connection",
     "get_git_status",
     "commit_to_git",
@@ -595,4 +592,7 @@
     "get_user_delegation_key",
     "refresh_sql_endpoint_metadata",
     "list_semantic_model_datasources",
+    "list_deployment_pipeline_stage_items",
+    "list_deployment_pipeline_stages",
+    "list_deployment_pipelines",
 ]

src/sempy_labs/_deployment_pipelines.py

@@ -9,6 +9,33 @@
 from uuid import UUID
 
 
+def resolve_deployment_pipeline_id(deployment_pipeline: str | UUID) -> UUID:
+    """
+    Obtains the Id for a given deployment pipeline.
+
+    Parameters
+    ----------
+    deployment_pipeline : str | uuid.UUID
+        The deployment pipeline name or ID.
+
+    Returns
+    -------
+    uuid.UUID
+        The deployment pipeline Id.
+    """
+
+    if _is_valid_uuid(deployment_pipeline):
+        return deployment_pipeline
+    else:
+        dfP = list_deployment_pipelines()
+        dfP_filt = dfP[dfP["Deployment Pipeline Name"] == deployment_pipeline]
+        if len(dfP_filt) == 0:
+            raise ValueError(
+                f"{icons.red_dot} The '{deployment_pipeline}' deployment pipeline is not valid."
+            )
+        return dfP_filt["Deployment Pipeline Id"].iloc[0]
+
+
 def list_deployment_pipelines() -> pd.DataFrame:
     """
     Shows a list of deployment pipelines the user can access.
@@ -63,8 +90,6 @@ def list_deployment_pipeline_stages(deployment_pipeline: str | UUID) -> pd.DataF
         A pandas dataframe showing the specified deployment pipeline stages.
     """
 
-    from sempy_labs._helper_functions import resolve_deployment_pipeline_id
-
     columns = {
         "Deployment Pipeline Stage Id": "string",
         "Deployment Pipeline Stage Name": "string",
@@ -126,8 +151,6 @@ def list_deployment_pipeline_stage_items(
         A pandas dataframe showing the supported items from the workspace assigned to the specified stage of the specified deployment pipeline.
     """
 
-    from sempy_labs._helper_functions import resolve_deployment_pipeline_id
-
     columns = {
         "Deployment Pipeline Stage Item Id": "string",
         "Deployment Pipeline Stage Item Name": "string",
@@ -181,3 +204,214 @@ def resolve_deployment_pipeline_stage_id(
             df = pd.concat([df, pd.DataFrame(new_data, index=[0])], ignore_index=True)
 
     return df
+
+
+def list_deployment_pipeline_role_assignments(
+    deployment_pipeline: str | UUID,
+) -> pd.DataFrame:
+    """
+    Shows the role assignments for the specified deployment pipeline.
+
+    This is a wrapper function for the following API: `Deployment Pipelines - List Deployment Pipeline Role Assignments <https://learn.microsoft.com/rest/api/fabric/core/deployment-pipelines/list-deployment-pipeline-role-assignments>`_.
+
+    Parameters
+    ----------
+    deployment_pipeline : str | uuid.UUID
+        The deployment pipeline name or ID.
+
+    Returns
+    -------
+    pandas.DataFrame
+        A pandas dataframe showing the role assignments for the specified deployment pipeline.
+    """
+
+    columns = {
+        "Role": "string",
+        "Principal Id": "string",
+        "Principal Type": "string",
+    }
+    df = _create_dataframe(columns=columns)
+
+    deployment_pipeline_id = resolve_deployment_pipeline_id(
+        deployment_pipeline=deployment_pipeline
+    )
+
+    responses = _base_api(
+        request=f"/v1/deploymentPipelines/{deployment_pipeline_id}/roleAssignments",
+        uses_pagination=True,
+        client="fabric_sp",
+    )
+
+    dfs = []
+
+    for r in responses:
+        for v in r.get("value", []):
+            principal = v.get("principal", {})
+            new_data = {
+                "Role": v.get("role"),
+                "Principal Id": principal.get("id"),
+                "Principal Type Name": principal.get("type"),
+            }
+            dfs.append(pd.DataFrame(new_data, index=[0]))
+
+    if dfs:
+        df = pd.concat(dfs, ignore_index=True)
+
+    return df
+
+
+def delete_deployment_pipeline(
+    deployment_pipeline: str | UUID,
+):
+    """
+    Deletes the specified deployment pipeline.
+
+    This is a wrapper function for the following API: `Deployment Pipelines - Delete Deployment Pipeline <https://learn.microsoft.com/rest/api/fabric/core/deployment-pipelines/delete-deployment-pipeline>`_.
+
+    Parameters
+    ----------
+    deployment_pipeline : str | uuid.UUID
+        The deployment pipeline name or ID.
+    """
+
+    deployment_pipeline_id = resolve_deployment_pipeline_id(
+        deployment_pipeline=deployment_pipeline
+    )
+
+    _base_api(
+        request=f"/v1/deploymentPipelines/{deployment_pipeline_id}",
+        method="delete",
+        client="fabric_sp",
+    )
+
+    print(
+        f"{icons.green_dot} The '{deployment_pipeline}' deployment pipeline has been deleted successfully."
+    )
+
+
+def list_deployment_pipeline_operations(
+    deployment_pipeline: str | UUID,
+) -> pd.DataFrame:
+    """
+    Shows the operations for the specified deployment pipeline.
+
+    This is a wrapper function for the following API: `Deployment Pipelines - List Deployment Pipeline Operations <https://learn.microsoft.com/rest/api/fabric/core/deployment-pipelines/list-deployment-pipeline-operations>`_.
+
+    Parameters
+    ----------
+    deployment_pipeline : str | uuid.UUID
+        The deployment pipeline name or ID.
+
+    Returns
+    -------
+    pandas.DataFrame
+        A pandas dataframe showing the operations for the specified deployment pipeline.
+    """
+
+    columns = {
+        "Operation Id": "string",
+        "Type": "string",
+        "Status": "string",
+        "Last Updated Time": "string",
+        "Execution Start Time": "datetime_coerce",
+        "Execution End Time": "datetime_coerce",
+        "Source Stage Id": "string",
+        "Target Stage Id": "string",
+        "Note": "string",
+        "New Items Count": "int",
+        "Different Items Count": "int",
+        "No Difference Items Count": "int",
+        "Performed By Id": "string",
+        "Performed By Type": "string",
+    }
+    df = _create_dataframe(columns=columns)
+
+    deployment_pipeline_id = resolve_deployment_pipeline_id(
+        deployment_pipeline=deployment_pipeline
+    )
+
+    responses = _base_api(
+        request=f"/v1/deploymentPipelines/{deployment_pipeline_id}/operations",
+        uses_pagination=True,
+        client="fabric_sp",
+    )
+
+    dfs = []
+    for r in responses:
+        for v in r.get("value", []):
+            p = v.get("preDeploymentDiffInformation", {})
+            new_data = {
+                "Operation Id": v.get("id"),
+                "Type": v.get("type"),
+                "Status": v.get("status"),
+                "Last Updated Time": v.get("lastUpdatedTime"),
+                "Execution Start Time": v.get("executionStartTime"),
+                "Execution End Time": v.get("executionEndTime"),
+                "Source Stage Id": v.get("sourceStageId"),
+                "Target Stage Id": v.get("targetStageId"),
+                "Note": v.get("note", {}).get("content"),
+                "New Items Count": p.get("newItemsCount"),
+                "Different Items Count": p.get("differentItemsCount"),
+                "No Difference Items Count": p.get("noDifferenceItemsCount"),
+                "Performed By Id": v.get("performedBy", {}).get("id"),
+                "Performed By Type": v.get("performedBy", {}).get("type"),
+            }
+            dfs.append(pd.DataFrame(new_data, index=[0]))
+
+    if dfs:
+        df = pd.concat(dfs, ignore_index=True)
+        _update_dataframe_datatypes(dataframe=df, column_map=columns)
+
+    return df
+
+
+def unassign_workspace_from_stage(
+    deployment_pipeline: str | UUID,
+    stage: str | UUID,
+):
+    """
+    Unassigns the workspace from the specified stage of the specified deployment pipeline.
+
+    This is a wrapper function for the following API: `Deployment Pipelines - Unassign Workspace From Stage <https://learn.microsoft.com/rest/api/fabric/core/deployment-pipelines/unassign-workspace-from-stage>`_.
+
+    Parameters
+    ----------
+    deployment_pipeline : str | uuid.UUID
+        The deployment pipeline name or ID.
+    stage : str | uuid.UUID
+        The deployment pipeline stage name or ID.
+    """
+
+    deployment_pipeline_id = resolve_deployment_pipeline_id(
+        deployment_pipeline=deployment_pipeline
+    )
+
+    def resolve_deployment_pipeline_stage_id(
+        deployment_pipeline_id: UUID, stage: str | UUID
+    ):
+
+        dfPS = list_deployment_pipeline_stages(
+            deployment_pipeline=deployment_pipeline_id
+        )
+
+        if _is_valid_uuid(stage):
+            dfPS_filt = dfPS[dfPS["Deployment Pipeline Stage Id"] == stage]
+        else:
+            dfPS_filt = dfPS[dfPS["Deployment Pipeline Stage Name"] == stage]
+        if dfPS.empty:
+            raise ValueError(
+                f"{icons.red_dot} The '{stage}' stage does not exist within the '{deployment_pipeline}' deployment pipeline."
+            )
+        return dfPS_filt["Deployment Pipeline Stage Id"].iloc[0]
+
+    stage_id = resolve_deployment_pipeline_stage_id(deployment_pipeline_id, stage)
+
+    _base_api(
+        request=f"/v1/deploymentPipelines/{deployment_pipeline_id}/stages/{stage_id}/unassignWorkspace",
+        method="post",
+        client="fabric_sp",
+    )
+
+    print(
+        f"{icons.green_dot} The workspace has been unassigned from the '{stage}' stage of the '{deployment_pipeline}' deployment pipeline successfully."
+    )

src/sempy_labs/_helper_functions.py

@@ -1311,36 +1311,6 @@ def pagination(client, response):
     return responses
 
 
-def resolve_deployment_pipeline_id(deployment_pipeline: str | UUID) -> UUID:
-    """
-    Obtains the Id for a given deployment pipeline.
-
-    Parameters
-    ----------
-    deployment_pipeline : str | uuid.UUID
-        The deployment pipeline name or ID.
-
-    Returns
-    -------
-    uuid.UUID
-        The deployment pipeline Id.
-    """
-
-    from sempy_labs._deployment_pipelines import list_deployment_pipelines
-
-    if _is_valid_uuid(deployment_pipeline):
-        return deployment_pipeline
-    else:
-
-        dfP = list_deployment_pipelines()
-        dfP_filt = dfP[dfP["Deployment Pipeline Name"] == deployment_pipeline]
-        if len(dfP_filt) == 0:
-            raise ValueError(
-                f"{icons.red_dot} The '{deployment_pipeline}' deployment pipeline is not valid."
-            )
-        return dfP_filt["Deployment Pipeline Id"].iloc[0]
-
-
 class FabricTokenCredential(TokenCredential):
 
     def get_token(

src/sempy_labs/_sql_endpoints.py

@@ -152,25 +152,33 @@ def refresh_sql_endpoint_metadata(
     df = pd.json_normalize(result)
 
     # Extract error code and message, set to None if no error
-    df['Error Code'] = df.get('error.errorCode', None)
-    df['Error Message'] = df.get('error.message', None)
+    df["Error Code"] = df.get("error.errorCode", None)
+    df["Error Message"] = df.get("error.message", None)
 
     # Friendly column renaming
-    df.rename(columns={
-        'tableName': 'Table Name',
-        'startDateTime': 'Start Time',
-        'endDateTime': 'End Time',
-        'status': 'Status',
-        'lastSuccessfulSyncDateTime': 'Last Successful Sync Time'
-    }, inplace=True)
+    df.rename(
+        columns={
+            "tableName": "Table Name",
+            "startDateTime": "Start Time",
+            "endDateTime": "End Time",
+            "status": "Status",
+            "lastSuccessfulSyncDateTime": "Last Successful Sync Time",
+        },
+        inplace=True,
+    )
 
     # Drop the original 'error' column if present
-    df.drop(columns=[col for col in ['error'] if col in df.columns], inplace=True)
+    df.drop(columns=[col for col in ["error"] if col in df.columns], inplace=True)
 
     # Optional: Reorder columns
     column_order = [
-        'Table Name', 'Status', 'Start Time', 'End Time',
-        'Last Successful Sync Time', 'Error Code', 'Error Message'
+        "Table Name",
+        "Status",
+        "Start Time",
+        "End Time",
+        "Last Successful Sync Time",
+        "Error Code",
+        "Error Message",
     ]
     df = df[column_order]
 

src/sempy_labs/admin/__init__.py

@@ -6,7 +6,7 @@
     add_user_to_workspace,
     delete_user_from_workspace,
     restore_deleted_workspace,
-    list_orphaned_workspaces
+    list_orphaned_workspaces,
 )
 from sempy_labs.admin._artifacts import (
     list_unused_artifacts,