Skip to content

Deployment Pipelines

add_deployment_pipeline_role_assignment(pipeline, user_uuid, user_type='User', role='Admin', *, df=True)

Adds a permission to a deployment pipeline for a user.

Parameters:

Name Type Description Default
pipeline str

The ID or name of the deployment pipeline.

required
user_uuid str

The UUID of the user.

required
user_type str

The type of user (options: User, Group, ServicePrincipal, ServicePrincipalProfile).

'User'
role str

The role to assign (options: admin, member, contributor, viewer).

'Admin'
df Optional[bool]

If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.

True

Returns:

Type Description
Union[DataFrame, Dict[str, Any], None]

The role assignment details if successful.

Raises:

Type Description
ResourceNotFoundError

If the specified deployment pipeline is not found.

OptionNotAvailableError

If the user type or role is invalid.

Examples:

add_deployment_pipeline_role_assignment(
    '123e4567-e89b-12d3-a456-426614174000',
    'FefEFewf-feF-1234-5678-9abcdef01234', user_type='User', role='Admin'
)
Source code in src/pyfabricops/core/deployment_pipelines.py
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
@df
def add_deployment_pipeline_role_assignment(
    pipeline: str,
    user_uuid: str,
    user_type: Literal[
        'User', 'Group', 'ServicePrincipal', 'ServicePrincipalProfile'
    ] = 'User',
    role: Literal['Admin', 'Contributor', 'Member', 'Viewer'] = 'Admin',
    *,
    df: Optional[bool] = True,
) -> Union[DataFrame, Dict[str, Any], None]:
    """
    Adds a permission to a deployment pipeline for a user.

    Args:
        pipeline (str): The ID or name of the deployment pipeline.
        user_uuid (str): The UUID of the user.
        user_type (str): The type of user (options: User, Group, ServicePrincipal, ServicePrincipalProfile).
        role (str): The role to assign (options: admin, member, contributor, viewer).
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a list of dictionaries.

    Returns:
        (Union[DataFrame, Dict[str, Any], None]): The role assignment details if successful.

    Raises:
        ResourceNotFoundError: If the specified deployment pipeline is not found.
        OptionNotAvailableError: If the user type or role is invalid.

    Examples:
        ```python
        add_deployment_pipeline_role_assignment(
            '123e4567-e89b-12d3-a456-426614174000',
            'FefEFewf-feF-1234-5678-9abcdef01234', user_type='User', role='Admin'
        )
        ```
    """
    if user_type not in [
        'User',
        'Group',
        'ServicePrincipal',
        'ServicePrincipalProfile',
    ]:
        raise OptionNotAvailableError(
            f'Invalid user type: {user_type}. Must be one of: User, Group, ServicePrincipal, ServicePrincipalProfile'
        )
    if role not in ['Admin', 'Contributor', 'Member', 'Viewer']:
        raise OptionNotAvailableError(
            f'Invalid role: {role}. Must be one of: Admin, Contributor, Member, Viewer'
        )
    payload = {'principal': {'id': user_uuid, 'type': user_type}, 'role': role}

    return api_request(
        '/deploymentPipelines/'
        + resolve_deployment_pipeline(pipeline)
        + '/roleAssignments',
        payload=payload,
        method='post',
    )

assign_workspace_to_stage(workspace, pipeline, stage, df=True)

Assigns a workspace to a deployment pipeline stage.

Parameters:

Name Type Description Default
pipeline str

The name or ID of the deployment pipeline.

required
stage str

The name or ID of the stage within the deployment pipeline.

required
workspace str

The name or ID of the workspace to assign.

required
df Optional[bool]

If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.

True

Returns:

Type Description
Union[DataFrame, List[Dict[str, Any]], None]

The updated details of the deployment pipeline stage.

Examples:

assign_workspace_to_stage(
    workspace='My Workspace',
    pipeline='My Deployment Pipeline',
    stage='Development'
)
Source code in src/pyfabricops/core/deployment_pipelines.py
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def assign_workspace_to_stage(
    workspace: str,
    pipeline: str,
    stage: str,
    df: Optional[bool] = True,
) -> Union[DataFrame, List[Dict[str, Any]], None]:
    """
    Assigns a workspace to a deployment pipeline stage.

    Args:
        pipeline (str): The name or ID of the deployment pipeline.
        stage (str): The name or ID of the stage within the deployment pipeline.
        workspace (str): The name or ID of the workspace to assign.
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a list of dictionaries.

    Returns:
        (Union[DataFrame, List[Dict[str, Any]], None]): The updated details of the deployment pipeline stage.

    Examples:
        ```python
        assign_workspace_to_stage(
            workspace='My Workspace',
            pipeline='My Deployment Pipeline',
            stage='Development'
        )
        ```
    """
    pipeline_id = resolve_deployment_pipeline(pipeline)
    if not pipeline_id:
        return None
    stage_id = resolve_deployment_pipeline_stage(pipeline, stage)
    if not stage_id:
        return None
    workspace_id = resolve_workspace(workspace)
    if not workspace_id:
        return None

    return api_request(
        f'/deploymentPipelines/{pipeline_id}/stages/{stage_id}/assignWorkspace',
        method='POST',
        payload={'workspaceId': workspace_id},
    )

create_deployment_pipeline(display_name, stages, *, description=None, df=True)

Creates a new deployment pipeline.

Parameters:

Name Type Description Default
display_name str

The display name of the deployment pipeline.

required
stages List[Dict[str, Any]]

A list of stages for the deployment pipeline.

required
description Optional[str]

An optional description for the deployment pipeline.

None
df Optional[bool]

If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.

True

Returns:

Type Description
Union[DataFrame, List[Dict[str, Any]], None]

The details of the created deployment pipeline.

Examples:

create_deployment_pipeline(
    display_name='My Deployment Pipeline',
    stages=[
        {
            "displayName": "Development",
            "description": "Development stage description",
            "isPublic": false
        },
        {
            "displayName": "Test",
            "description": "Test stage description",
            "isPublic": false
        },
        {
            "displayName": "Production",
            "description": "Production stage description",
            "isPublic": true
        }
    ],
    description='This is my deployment pipeline',
    df=True
)
Source code in src/pyfabricops/core/deployment_pipelines.py
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
@df
def create_deployment_pipeline(
    display_name: str,
    stages: List[Dict[str, Any]],
    *,
    description: Optional[str] = None,
    df: Optional[bool] = True,
) -> Union[DataFrame, List[Dict[str, Any]], None]:
    """
    Creates a new deployment pipeline.

    Args:
        display_name (str): The display name of the deployment pipeline.
        stages (List[Dict[str, Any]]): A list of stages for the deployment pipeline.
        description (Optional[str]): An optional description for the deployment pipeline.
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a list of dictionaries.

    Returns:
        (Union[DataFrame, List[Dict[str, Any]], None]): The details of the created deployment pipeline.

    Examples:
        ```python
        create_deployment_pipeline(
            display_name='My Deployment Pipeline',
            stages=[
                {
                    "displayName": "Development",
                    "description": "Development stage description",
                    "isPublic": false
                },
                {
                    "displayName": "Test",
                    "description": "Test stage description",
                    "isPublic": false
                },
                {
                    "displayName": "Production",
                    "description": "Production stage description",
                    "isPublic": true
                }
            ],
            description='This is my deployment pipeline',
            df=True
        )
        ```
    """
    payload = {
        'displayName': display_name,
        'stages': stages,
    }
    if description:
        payload['description'] = description
    return api_request(
        '/deploymentPipelines',
        method='POST',
        payload=payload,
    )

delete_deployment_pipeline(pipeline)

Deletes a deployment pipeline.

Parameters:

Name Type Description Default
pipeline str

The name or ID of the deployment pipeline to delete.

required

Returns:

Type Description
None

None

Examples:

delete_deployment_pipeline('123e4567-e89b-12d3-a456-426614174000')
Source code in src/pyfabricops/core/deployment_pipelines.py
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
def delete_deployment_pipeline(
    pipeline: str,
) -> None:
    """
    Deletes a deployment pipeline.

    Args:
        pipeline (str): The name or ID of the deployment pipeline to delete.

    Returns:
        None

    Examples:
        ```python
        delete_deployment_pipeline('123e4567-e89b-12d3-a456-426614174000')
        ```
    """
    pipeline_id = resolve_deployment_pipeline(pipeline)
    if not pipeline_id:
        return None

    api_request(
        '/deploymentPipelines/' + pipeline_id,
        method='DELETE',
    )

delete_deployment_pipeline_role_assignment(pipeline, role_assignment_id)

Deletes a role assignment from a deployment pipeline.

Parameters:

Name Type Description Default
pipeline str

The name or ID of the deployment pipeline.

required
role_assignment_id str

The ID of the role assignment to delete.

required

Returns:

Type Description
None

None

Examples:

delete_deployment_pipeline_role_assignment(
    '123e4567-e89b-12d3-a456-426614174000',
    'role-assignment-uuid'
)
Source code in src/pyfabricops/core/deployment_pipelines.py
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
def delete_deployment_pipeline_role_assignment(
    pipeline: str,
    role_assignment_id: str,
) -> None:
    """
    Deletes a role assignment from a deployment pipeline.

    Args:
        pipeline (str): The name or ID of the deployment pipeline.
        role_assignment_id (str): The ID of the role assignment to delete.

    Returns:
        None

    Examples:
        ```python
        delete_deployment_pipeline_role_assignment(
            '123e4567-e89b-12d3-a456-426614174000',
            'role-assignment-uuid'
        )
        ```
    """
    pipeline_id = resolve_deployment_pipeline(pipeline)
    if not pipeline_id:
        return None

    api_request(
        '/deploymentPipelines/'
        + pipeline_id
        + '/roleAssignments/'
        + role_assignment_id,
        method='DELETE',
    )

deploy_stage_content(pipeline, source_stage, target_stage, *, items=None, note=None, options=None, df=True)

Deploys content to a specified stage in a deployment pipeline.

Parameters:

Name Type Description Default
pipeline str

The name or ID of the deployment pipeline.

required
source_stage str

The name or ID of the source stage within the deployment pipeline.

required
target_stage str

The name or ID of the target stage within the deployment pipeline.

required
items Optional[List[str]]

A list of item IDs to deploy. If not provided, all items will be deployed.

None
note Optional[str]

An optional note for the deployment.

None
options Optional[Dict[str, Any]]

Additional deployment options.

None
df Optional[bool]

If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.

True

Returns:

Type Description
Union[DataFrame, List[Dict[str, Any]], None]

The deployment result.

Examples:

deploy_stage_content(
    '123e4567-e89b-12d3-a456-426614174000',
    'develop',
    'staging',
)

deploy_stage_content(
    pipeline = '123e4567-e89b-12d3-a456-426614174000',
    source_stage = 'develop',
    target_stage = 'staging',
    items = [
        {
        "sourceItemId": "1a201f2a-d1d8-45c0-8c61-1676338517de",
        "itemType": "SemanticModel"
        },
        {
        "sourceItemId": "2d225191-65f8-4ec3-b77d-06100602b1f7",
        "itemType": "Report"
        }
    ],
    note = "Deploying selected items from develop to staging",
    options = {"allowCrossRegionDeployment": True}
)
Source code in src/pyfabricops/core/deployment_pipelines.py
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
@df
def deploy_stage_content(
    pipeline: str,
    source_stage: str,
    target_stage: str,
    *,
    items: Optional[List[Dict[str, Any]]] = None,
    note: Optional[str] = None,
    options: Optional[Dict[str, Any]] = None,
    df: Optional[bool] = True,
) -> Union[DataFrame, List[Dict[str, Any]], None]:
    """
    Deploys content to a specified stage in a deployment pipeline.

    Args:
        pipeline (str): The name or ID of the deployment pipeline.
        source_stage (str): The name or ID of the source stage within the deployment pipeline.
        target_stage (str): The name or ID of the target stage within the deployment pipeline.
        items (Optional[List[str]]): A list of item IDs to deploy. If not provided, all items will be deployed.
        note (Optional[str]): An optional note for the deployment.
        options (Optional[Dict[str, Any]]): Additional deployment options.
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a list of dictionaries.

    Returns:
        (Union[DataFrame, List[Dict[str, Any]], None]): The deployment result.

    Examples:
        ```python
        deploy_stage_content(
            '123e4567-e89b-12d3-a456-426614174000',
            'develop',
            'staging',
        )

        deploy_stage_content(
            pipeline = '123e4567-e89b-12d3-a456-426614174000',
            source_stage = 'develop',
            target_stage = 'staging',
            items = [
                {
                "sourceItemId": "1a201f2a-d1d8-45c0-8c61-1676338517de",
                "itemType": "SemanticModel"
                },
                {
                "sourceItemId": "2d225191-65f8-4ec3-b77d-06100602b1f7",
                "itemType": "Report"
                }
            ],
            note = "Deploying selected items from develop to staging",
            options = {"allowCrossRegionDeployment": True}
        )
        ```
    """
    pipeline_id = resolve_deployment_pipeline(pipeline)
    if not pipeline_id:
        return None
    source_stage_id = resolve_deployment_pipeline_stage(
        pipeline_id, source_stage
    )
    if not source_stage_id:
        return None
    target_stage_id = resolve_deployment_pipeline_stage(
        pipeline_id, target_stage
    )
    if not target_stage_id:
        return None

    payload = {
        'sourceStageId': source_stage_id,
        'targetStageId': target_stage_id,
    }

    if items:
        payload['items'] = items
    if note:
        payload['note'] = note
    if options:
        payload['options'] = options

    return api_request(
        '/deploymentPipelines/'
        + resolve_deployment_pipeline(pipeline)
        + '/deploy',
        method='POST',
        payload=payload,
        support_lro=True,
    )

get_deployment_pipeline(pipeline, df=True)

Returns the specified deployment pipeline.

Parameters:

Name Type Description Default
pipeline str

The name or ID of the deployment pipeline to retrieve.

required
df Optional[bool]

If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.

True

Returns:

Type Description
Union[DataFrame, List[Dict[str, Any]], None]

(Union[DataFrame, List[Dict[str, Any]], None]) The details of the deployment pipeline if found, otherwise None. If df=True, returns a DataFrame with flattened keys.

Examples:

get_deployment_pipeline('123e4567-e89b-12d3-a456-426614174000')
get_deployment_pipeline('MyProjectPipeline')
get_deployment_pipeline('MyProjectPipeline', df=False) # Returns as list
Source code in src/pyfabricops/core/deployment_pipelines.py
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@df
def get_deployment_pipeline(
    pipeline: str, df: Optional[bool] = True
) -> Union[DataFrame, List[Dict[str, Any]], None]:
    """
    Returns the specified deployment pipeline.

    Args:
        pipeline (str): The name or ID of the deployment pipeline to retrieve.
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
                If False, returns a list of dictionaries.

    Returns:
        (Union[DataFrame, List[Dict[str, Any]], None]) The details of the deployment pipeline if found, otherwise None. If `df=True`, returns a DataFrame with flattened keys.

    Examples:
        ```python
        get_deployment_pipeline('123e4567-e89b-12d3-a456-426614174000')
        get_deployment_pipeline('MyProjectPipeline')
        get_deployment_pipeline('MyProjectPipeline', df=False) # Returns as list
        ```
    """
    return api_request(
        '/deploymentPipelines/' + resolve_deployment_pipeline(pipeline)
    )

get_deployment_pipeline_id(pipeline_name)

Retrieves the ID of a deployment pipeline by its name.

Parameters:

Name Type Description Default
pipeline_name str

The name of the deployment pipeline.

required

Returns:

Type Description
Union[str, None]

str | None: The ID of the deployment pipeline if found, otherwise None.

Source code in src/pyfabricops/core/deployment_pipelines.py
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def get_deployment_pipeline_id(pipeline_name: str) -> Union[str, None]:
    """
    Retrieves the ID of a deployment pipeline by its name.

    Args:
        pipeline_name (str): The name of the deployment pipeline.

    Returns:
        str | None: The ID of the deployment pipeline if found, otherwise None.
    """
    pipelines = list_deployment_pipelines(df=False)
    for _pipeline in pipelines:
        if _pipeline['displayName'] == pipeline_name:
            return _pipeline['id']

    logger.warning(f"Deployment pipeline '{pipeline_name}' not found.")
    return None

get_deployment_pipeline_operation(pipeline, operation_id, *, df=True)

Gets a specific operation for a deployment pipeline.

Parameters:

Name Type Description Default
pipeline str

The name or ID of the deployment pipeline.

required
operation_id str

The ID of the operation to retrieve.

required
df Optional[bool]

If True or not provided, returns a DataFrame with flattened keys. If False, returns a dictionary.

True

Returns:

Type Description
Union[DataFrame, Dict[str, Any], None]

The requested deployment pipeline operation.

Source code in src/pyfabricops/core/deployment_pipelines.py
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
@df
def get_deployment_pipeline_operation(
    pipeline: str,
    operation_id: str,
    *,
    df: Optional[bool] = True,
) -> Union[DataFrame, Dict[str, Any], None]:
    """
    Gets a specific operation for a deployment pipeline.

    Args:
        pipeline (str): The name or ID of the deployment pipeline.
        operation_id (str): The ID of the operation to retrieve.
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a dictionary.

    Returns:
        (Union[DataFrame, Dict[str, Any], None]): The requested deployment pipeline operation.
    """
    pipeline_id = resolve_deployment_pipeline(pipeline)
    if not pipeline_id:
        return None

    return api_request(
        '/deploymentPipelines/' + pipeline_id + '/operations/' + operation_id,
        method='GET',
    )

list_deployment_pipeline_operations(pipeline, *, df=True)

Lists all operations for a deployment pipeline.

Parameters:

Name Type Description Default
pipeline str

The name or ID of the deployment pipeline to list operations for.

required
df Optional[bool]

If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.

True

Returns:

Type Description
Union[DataFrame, List[Dict[str, Any]], None]

A list of deployment pipeline operations.

Source code in src/pyfabricops/core/deployment_pipelines.py
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
@df
def list_deployment_pipeline_operations(
    pipeline: str,
    *,
    df: Optional[bool] = True,
) -> Union[DataFrame, List[Dict[str, Any]], None]:
    """
    Lists all operations for a deployment pipeline.

    Args:
        pipeline (str): The name or ID of the deployment pipeline to list operations for.
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a list of dictionaries.

    Returns:
        (Union[DataFrame, List[Dict[str, Any]], None]): A list of deployment pipeline operations.
    """
    pipeline_id = resolve_deployment_pipeline(pipeline)
    if not pipeline_id:
        return None

    return api_request(
        '/deploymentPipelines/' + pipeline_id + '/operations',
        method='GET',
        support_pagination=True,
    )

list_deployment_pipeline_role_assignments(pipeline, *, df=True)

Lists all role assignments for a deployment pipeline.

Parameters:

Name Type Description Default
pipeline str

The name or ID of the deployment pipeline to list role assignments for.

required
df Optional[bool]

If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.

True

Returns:

Type Description
Union[DataFrame, List[Dict[str, Any]], None]

A list of role assignments.

Examples:

list_deployment_pipeline_role_assignments('123e4567-e89b-12d3-a456-426614174000')
Source code in src/pyfabricops/core/deployment_pipelines.py
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
@df
def list_deployment_pipeline_role_assignments(
    pipeline: str,
    *,
    df: Optional[bool] = True,
) -> Union[DataFrame, List[Dict[str, Any]], None]:
    """
    Lists all role assignments for a deployment pipeline.

    Args:
        pipeline (str): The name or ID of the deployment pipeline to list role assignments for.
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a list of dictionaries.

    Returns:
        (Union[DataFrame, List[Dict[str, Any]], None]): A list of role assignments.

    Examples:
        ```python
        list_deployment_pipeline_role_assignments('123e4567-e89b-12d3-a456-426614174000')
        ```
    """
    return api_request(
        '/deploymentPipelines/'
        + resolve_deployment_pipeline(pipeline)
        + '/roleAssignments',
        method='GET',
        support_pagination=True,
    )

list_deployment_pipelines(df=True)

List deployment pipelines

Parameters:

Name Type Description Default
df Optional[bool]

If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.

True

Returns:

Type Description
Union[DataFrame, List[Dict[str, Any]], None]

A list of deployment pipelines.

Examples:

list_deployment_pipelines()
Source code in src/pyfabricops/core/deployment_pipelines.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@df
def list_deployment_pipelines(
    df: Optional[bool] = True,
) -> Union[DataFrame, List[Dict[str, Any]], None]:
    """
    List deployment pipelines

    Args:
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a list of dictionaries.

    Returns:
        (Union[DataFrame, List[Dict[str, Any]], None]): A list of deployment pipelines.

    Examples:
        ```python
        list_deployment_pipelines()
        ```
    """
    return api_request(
        '/deploymentPipelines',
        support_pagination=True,
    )

resolve_deployment_pipeline(pipeline)

Resolves a deployment pipeline to its ID.

Parameters:

Name Type Description Default
pipeline str

The name or id of the deployment pipeline.

required

Returns:

Type Description
Union[str, None]

str | None: The ID of the deployment pipeline if found, otherwise None.

Source code in src/pyfabricops/core/deployment_pipelines.py
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def resolve_deployment_pipeline(pipeline: str) -> Union[str, None]:
    """
    Resolves a deployment pipeline to its ID.

    Args:
        pipeline (str): The name or id of the deployment pipeline.

    Returns:
        str | None: The ID of the deployment pipeline if found, otherwise None.
    """
    if is_valid_uuid(pipeline):
        return pipeline
    else:
        return get_deployment_pipeline_id(pipeline)

resolve_deployment_pipeline_stage(pipeline, stage_name)

Resolves a deployment pipeline stage to its ID.

Parameters:

Name Type Description Default
pipeline str

The name or id of the deployment pipeline.

required
stage_name str

The name of the deployment pipeline stage.

required

Returns:

Type Description
Union[str, None]

str | None: The ID of the deployment pipeline stage if found, otherwise None.

Source code in src/pyfabricops/core/deployment_pipelines.py
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def resolve_deployment_pipeline_stage(
    pipeline: str, stage_name: str
) -> Union[str, None]:
    """
    Resolves a deployment pipeline stage to its ID.

    Args:
        pipeline (str): The name or id of the deployment pipeline.
        stage_name (str): The name of the deployment pipeline stage.

    Returns:
        str | None: The ID of the deployment pipeline stage if found, otherwise None.
    """
    pipeline_id = resolve_deployment_pipeline(pipeline)
    if not pipeline_id:
        return None
    pipeline_details = get_deployment_pipeline(pipeline, df=False)
    for stage in pipeline_details.get('stages', []):
        if stage['displayName'] == stage_name:
            return stage['id']

    logger.warning(
        f"Stage '{stage_name}' not found in deployment pipeline '{pipeline}'."
    )
    return None

unassign_workspace_to_stage(pipeline, stage)

Unassigns a workspace from a deployment pipeline stage.

Parameters:

Name Type Description Default
pipeline str

The name or ID of the deployment pipeline.

required
stage str

The name or ID of the stage within the deployment pipeline.

required
workspace str

The name or ID of the workspace to unassign.

required
Source code in src/pyfabricops/core/deployment_pipelines.py
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def unassign_workspace_to_stage(
    pipeline: str,
    stage: str,
) -> Union[DataFrame, List[Dict[str, Any]], None]:
    """
    Unassigns a workspace from a deployment pipeline stage.

    Args:
        pipeline (str): The name or ID of the deployment pipeline.
        stage (str): The name or ID of the stage within the deployment pipeline.
        workspace (str): The name or ID of the workspace to unassign.
    """
    pipeline_id = resolve_deployment_pipeline(pipeline)
    if not pipeline_id:
        return None
    stage_id = resolve_deployment_pipeline_stage(pipeline, stage)
    if not stage_id:
        return None

    return api_request(
        f'/deploymentPipelines/{pipeline_id}/stages/{stage_id}/unassignWorkspace',
        method='POST',
    )

update_deployment_pipeline(pipeline, display_name=None, description=None, *, df=True)

Updates a deployment pipeline's details.

Parameters:

Name Type Description Default
pipeline str

The name or ID of the deployment pipeline to update.

required
display_name Optional[str]

The new display name for the deployment pipeline.

None
description Optional[str]

The new description for the deployment pipeline.

None
df Optional[bool]

If True or not provided, returns a DataFrame with flattened keys. If False, returns a dictionary.

True

Returns:

Type Description
Union[DataFrame, Dict[str, Any], None]

The updated deployment pipeline details.

Examples:

updated_pipeline = update_deployment_pipeline(
    pipeline="my-pipeline",
    display_name="My Updated Pipeline",
    description="This is an updated description."
)
Source code in src/pyfabricops/core/deployment_pipelines.py
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
@df
def update_deployment_pipeline(
    pipeline: str,
    display_name: Optional[str] = None,
    description: Optional[str] = None,
    *,
    df: Optional[bool] = True,
) -> Union[DataFrame, Dict[str, Any], None]:
    """
    Updates a deployment pipeline's details.

    Args:
        pipeline (str): The name or ID of the deployment pipeline to update.
        display_name (Optional[str]): The new display name for the deployment pipeline.
        description (Optional[str]): The new description for the deployment pipeline.
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a dictionary.

    Returns:
        (Union[DataFrame, Dict[str, Any], None]): The updated deployment pipeline details.

    Examples:
        ```python
        updated_pipeline = update_deployment_pipeline(
            pipeline="my-pipeline",
            display_name="My Updated Pipeline",
            description="This is an updated description."
        )
        ```
    """
    pipeline_id = resolve_deployment_pipeline(pipeline)
    if not pipeline_id:
        return None

    payload = {}
    if display_name:
        payload['displayName'] = display_name
    if description:
        payload['description'] = description

    return api_request(
        '/deploymentPipelines/' + pipeline_id,
        method='PATCH',
        payload=payload,
    )

update_deployment_pipeline_stage(pipeline, stage, display_name=None, description=None, is_public=None)

Updates a deployment pipeline stage's details.

Parameters:

Name Type Description Default
pipeline str

The name or ID of the deployment pipeline.

required
stage str

The name or ID of the stage within the deployment pipeline.

required
display_name Optional[str]

The new display name for the stage.

None
description Optional[str]

The new description for the stage.

None
is_public Optional[bool]

Whether the stage is public or not.

None

Returns:

Type Description
Union[DataFrame, Dict[str, Any], None]

The updated deployment pipeline stage details.

Source code in src/pyfabricops/core/deployment_pipelines.py
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
def update_deployment_pipeline_stage(
    pipeline: str,
    stage: str,
    display_name: Optional[str] = None,
    description: Optional[str] = None,
    is_public: Optional[bool] = None,
) -> Union[DataFrame, Dict[str, Any], None]:
    """
    Updates a deployment pipeline stage's details.

    Args:
        pipeline (str): The name or ID of the deployment pipeline.
        stage (str): The name or ID of the stage within the deployment pipeline.
        display_name (Optional[str]): The new display name for the stage.
        description (Optional[str]): The new description for the stage.
        is_public (Optional[bool]): Whether the stage is public or not.

    Returns:
        (Union[DataFrame, Dict[str, Any], None]): The updated deployment pipeline stage details.
    """
    pipeline_id = resolve_deployment_pipeline(pipeline)
    if not pipeline_id:
        return None
    stage_id = resolve_deployment_pipeline_stage(pipeline, stage)
    if not stage_id:
        return None

    payload = {}
    if display_name:
        payload['displayName'] = display_name
    if description:
        payload['description'] = description
    if is_public is not None:
        payload['isPublic'] = is_public

    return api_request(
        f'/deploymentPipelines/{pipeline_id}/stages/{stage_id}',
        method='PATCH',
        payload=payload,
    )