msfabric-devops 0.1.9

Python package to interact with Microsoft Fabric items.

msfabric-devops

A Python package to interact with Microsoft Fabric objects, providing functionality to manage workspaces, semantic models, and items through the Fabric REST API.

Table of Contents

• Installation
• Requirements
• Authentication
  • get_access_token
• Workspaces
  • get_workspaces
  • get_workspace_by_id
  • get_workspaces_by_name
  • create_workspace
  • delete_workspace
• Items
  • get_items
  • get_item_by_id
  • get_items_by_name
  • get_item_definition_by_id
  • import_item
  • delete_item_by_id
• Internal API Functions
  • invoke_fabric_api_request
• Complete Example
• License
• Author

Installation

pip install msfabric-devops

Requirements

• Python >= 3.9
• Azure AD service principal credentials (tenant ID, client ID, client secret)
• azure-identity package

Authentication

get_access_token(tenant_id=None, client_id=None, client_secret=None) -> str

Authenticates using a service principal and returns an access token for the Fabric REST API.

Parameters:

• tenant_id (str): Azure AD Tenant ID.
• client_id (str): Azure AD Application (client) ID.
• client_secret (str): Azure AD Client Secret.

Returns:

• str: A valid access token string for authenticating Fabric API requests.

Example:

from msfabric_devops import get_access_token

token = get_access_token(
    tenant_id="your-tenant-id",
    client_id="your-client-id",
    client_secret="your-client-secret"
)

Workspaces

get_workspaces(token) -> list[dict]

Retrieves all Fabric workspaces accessible to the authenticated user.

Parameters:

• token (str): Access token from get_access_token().

Returns:

• list[dict]: List of workspace dictionaries containing workspace details (id, displayName, type, etc.).

Example:

from msfabric_devops import get_access_token, get_workspaces

token = get_access_token()
workspaces = get_workspaces(token)
for workspace in workspaces:
    print(workspace["displayName"])

get_workspace_by_id(token, workspace_id) -> dict

Retrieves a specific workspace by its ID.

Parameters:

• token (str): Access token from get_access_token().
• workspace_id (str): The unique identifier of the workspace.

Returns:

• dict: Workspace dictionary containing workspace details. Returns empty dict if not found.

Example:

from msfabric_devops import get_access_token, get_workspace_by_id

token = get_access_token()
workspace = get_workspace_by_id(token, "workspace-id-here")
print(workspace["displayName"])

get_workspaces_by_name(token, workspace_name) -> list[dict]

Retrieves all workspaces matching a specific display name.

Parameters:

• token (str): Access token from get_access_token().
• workspace_name (str): The display name to search for (exact match).

Returns:

• list[dict]: List of workspace dictionaries matching the name.

Example:

from msfabric_devops import get_access_token, get_workspaces_by_name

token = get_access_token()
workspaces = get_workspaces_by_name(token, "My Workspace")

create_workspace(token, workspace_name) -> dict

Creates a new Fabric workspace with the specified name.

Parameters:

• token (str): Access token from get_access_token().
• workspace_name (str): Display name for the new workspace.

Returns:

• dict: Created workspace dictionary containing workspace details. Returns None if workspace already exists (with a warning message).

Example:

from msfabric_devops import get_access_token, create_workspace

token = get_access_token()
workspace = create_workspace(token, "New Workspace")

delete_workspace(token, workspace_id) -> None

Deletes a Fabric workspace by its ID.

Parameters:

• token (str): Access token from get_access_token().
• workspace_id (str): The unique identifier of the workspace to delete.

Returns:

• None: No return value on success. Raises exception on error.

Example:

from msfabric_devops import get_access_token, delete_workspace

token = get_access_token()
delete_workspace(token, "workspace-id-here")

Items

get_items(token, workspace_id) -> list[dict]

Retrieves all items in a specified Fabric workspace.

Parameters:

• token (str): Access token from get_access_token().
• workspace_id (str): The unique identifier of the workspace.

Returns:

• list[dict]: List of item dictionaries containing item details (id, displayName, type, etc.).

Example:

from msfabric_devops import get_access_token, get_items

token = get_access_token()
items = get_items(token, "workspace-id-here")
for item in items:
    print(f"{item['displayName']} ({item['type']})")

get_item_by_id(token, workspace_id, item_id) -> dict

Retrieves a specific item by its ID within a workspace.

Parameters:

• token (str): Access token from get_access_token().
• workspace_id (str): The unique identifier of the workspace.
• item_id (str): The unique identifier of the item.

Returns:

• dict: Item dictionary containing item details.

Example:

from msfabric_devops import get_access_token, get_item_by_id

token = get_access_token()
item = get_item_by_id(token, "workspace-id", "item-id")
print(item["displayName"])

get_items_by_name(token, workspace_id, item_name) -> list[dict]

Retrieves all items matching a specific display name within a workspace.

Parameters:

• token (str): Access token from get_access_token().
• workspace_id (str): The unique identifier of the workspace.
• item_name (str): The display name to search for (exact match).

Returns:

• list[dict]: List of item dictionaries matching the name.

Example:

from msfabric_devops import get_access_token, get_items_by_name

token = get_access_token()
items = get_items_by_name(token, "workspace-id", "My Item")

get_item_definition_by_id(token, workspace_id, item_id, output_dir=None, format=None) -> dict

Exports an item definition from a Fabric workspace. Optionally saves the definition files to a local directory.

Parameters:

• token (str): Access token from get_access_token().
• workspace_id (str): The unique identifier of the workspace.
• item_id (str): The unique identifier of the item.
• output_dir (str, optional): Local directory path where definition files should be saved. If provided, files are decoded from Base64 and written to disk. Defaults to None (no files saved).
• format (str, optional): Export format (e.g., 'PBIP'). Defaults to None.

Returns:

• dict: Response dictionary containing the definition with parts array. Each part includes path, payload (Base64 encoded), and payloadType.

Example:

from msfabric_devops import get_access_token, get_item_definition_by_id

token = get_access_token()
definition = get_item_definition_by_id(
    token,
    "workspace-id",
    "item-id",
    output_dir="./output"
)

import_item(token, workspace_id, path, item_properties=None, skip_if_exists=False, retain_roles=False) -> dict

Imports a Fabric item (semantic model or report) from a local PBIP folder into a Fabric workspace. Supports both .pbism (semantic models) and .pbir (reports) files.

Parameters:

• token (str): Access token from get_access_token().
• workspace_id (str): The unique identifier of the target workspace.
• path (str): Local folder path containing the PBIP export. Must contain either a .pbism or .pbir file.
• item_properties (dict, optional): Dictionary to override item properties:
  • displayName (str): Override the display name
  • semanticModelId (str): Required when importing reports that use byPath connections to semantic models
  • type (str): Override the item type (usually auto-detected)
• skip_if_exists (bool, optional): If True, does not update the definition if an item with the same name and type already exists. Defaults to False.
• retain_roles (bool, optional): If True, preserves existing RLS (Row-Level Security) roles from the published model when updating. This option:
  • Fetches the current model definition
  • Extracts role definitions from definition/roles/*.tmdl files
  • Merges them with the new definition
  • Updates definition/model.tmdl to include role references Defaults to False.

Returns:

• dict: Dictionary containing the imported/updated item details with keys: id, displayName, type.

Behavior:

• Item Detection: Automatically detects item type based on .pbism (SemanticModel) or .pbir (Report) files
• File Processing: Processes all files in the folder except:
  • Files starting with item.
  • Files with .abf extension
  • Files in .pbi directory
• Report Connections: For reports using byPath connections, you must provide item_properties.semanticModelId to convert to byConnection format
• Create vs Update: Creates a new item if none exists with the same name and type, otherwise updates the existing item

Example:

from msfabric_devops import get_access_token,import_item

token = get_access_token()

# Import a semantic model
result = import_item(
    token,
    "workspace-id",
    r"C:\path\to\semantic-model-pbip",
    item_properties={"displayName": "My Semantic Model"},
    retain_roles=True
)

# Import a report connected to a semantic model
result = import_item(
    token,
    "workspace-id",
    r"C:\path\to\report-pbip",
    item_properties={
        "displayName": "My Report",
        "semanticModelId": "existing-semantic-model-id"
    }
)

delete_item_by_id(token, workspace_id, item_id) -> None

Deletes a Fabric item by its ID.

Parameters:

• token (str): Access token from get_access_token().
• workspace_id (str): The unique identifier of the workspace.
• item_id (str): The unique identifier of the item to delete.

Returns:

• None: No return value on success. Raises exception on error.

Example:

from msfabric_devops.authenticate import get_access_token
from msfabric_devops import delete_item_by_id

token = get_access_token()
delete_item_by_id(token, "workspace-id", "item-id")

Internal API Functions

invoke_fabric_api_request(uri, token=None, method="GET", body=None, content_type="application/json; charset=utf-8", timeout_sec=240, retry_count=0, api_url=None) -> dict | list | None

Low-level function to make requests to the Fabric REST API. Handles authentication, error handling, throttling, and long-running operations.

Parameters:

• uri (str): API endpoint URI (relative to base API URL).
• token (str, optional): Bearer token for authentication.
• method (str, optional): HTTP method ("GET", "POST", "DELETE", etc.). Defaults to "GET".
• body (dict | list | str, optional): Request body. If dict/list, sent as JSON; if str, sent as raw data.
• content_type (str, optional): Content-Type header. Defaults to "application/json; charset=utf-8".
• timeout_sec (int, optional): Request timeout in seconds. Defaults to 240.
• retry_count (int, optional): Internal retry counter for throttling. Defaults to 0.
• api_url (str, optional): Base API URL. Defaults to "https://api.fabric.microsoft.com/v1".

Returns:

• dict | list | None: Parsed JSON response. Returns None for successful LRO operations with no result.

Features:

• Long-Running Operations (LRO): Automatically polls Location header for 202 responses
• Throttling: Retries up to 3 times on 429 (Too Many Requests) with exponential backoff
• Error Handling: Raises exceptions for API errors and network issues
• JSON Parsing: Automatically extracts value field from responses if present

Complete Example

from msfabric_devops import (
    get_access_token,
    get_workspaces,
    create_workspace,
    get_items,
    import_item
)

# Authenticate
token = get_access_token()

# List all workspaces
workspaces = get_workspaces(token)
print(f"Found {len(workspaces)} workspaces")

# Create a new workspace
workspace = create_workspace(token, "My New Workspace")
workspace_id = workspace["id"]

# List items in the workspace
items = get_items(token, workspace_id)
print(f"Found {len(items)} items")

# Import a semantic model
result = import_item(
    token,
    workspace_id,
    r"C:\path\to\pbip\export",
    item_properties={"displayName": "Published Model"},
    retain_roles=True
)
print(f"Imported: {result['displayName']} (ID: {result['id']})")

License

MIT

Author

Hugo Salaun ([e-mail])