🔌 Fabric REST APIs - Programmatic Platform Management¶

import requests
from azure.identity import ClientSecretCredential

# Authenticate with service principal
credential = ClientSecretCredential(
    tenant_id="your-tenant-id",
    client_id="your-client-id",
    client_secret="your-client-secret"
)

token = credential.get_token("https://api.fabric.microsoft.com/.default")

headers = {
    "Authorization": f"Bearer {token.token}",
    "Content-Type": "application/json"
}

# List workspaces
response = requests.get(
    "https://api.fabric.microsoft.com/v1/workspaces",
    headers=headers
)
workspaces = response.json()["value"]

from azure.identity import ManagedIdentityCredential

# System-assigned managed identity (from Azure VM, App Service, ADF)
credential = ManagedIdentityCredential()
token = credential.get_token("https://api.fabric.microsoft.com/.default")

# User-assigned managed identity
credential = ManagedIdentityCredential(
    client_id="user-assigned-mi-client-id"
)

def create_workspace(headers, display_name, capacity_id, description=""):
    """Create a new Fabric workspace and assign it to a capacity."""
    payload = {
        "displayName": display_name,
        "description": description,
        "capacityId": capacity_id
    }

    response = requests.post(
        "https://api.fabric.microsoft.com/v1/workspaces",
        headers=headers,
        json=payload
    )
    response.raise_for_status()
    return response.json()

def add_workspace_member(headers, workspace_id, principal_id, role="Member"):
    """Add a user or service principal to a workspace with a specific role."""
    payload = {
        "principal": {
            "id": principal_id,
            "type": "User"  # or "ServicePrincipal", "Group"
        },
        "role": role  # Admin, Member, Contributor, Viewer
    }

    response = requests.post(
        f"https://api.fabric.microsoft.com/v1/workspaces/{workspace_id}/roleAssignments",
        headers=headers,
        json=payload
    )
    response.raise_for_status()

def create_lakehouse(headers, workspace_id, lakehouse_name):
    """Create a new Lakehouse item in a workspace."""
    payload = {
        "displayName": lakehouse_name,
        "type": "Lakehouse"
    }

    response = requests.post(
        f"https://api.fabric.microsoft.com/v1/workspaces/{workspace_id}/items",
        headers=headers,
        json=payload
    )
    response.raise_for_status()
    return response.json()

def list_workspace_items(headers, workspace_id, item_type=None):
    """List all items in a workspace, optionally filtered by type."""
    url = f"https://api.fabric.microsoft.com/v1/workspaces/{workspace_id}/items"
    if item_type:
        url += f"?type={item_type}"

    items = []
    while url:
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        data = response.json()
        items.extend(data["value"])
        url = data.get("continuationUri")

    return items

import base64

def deploy_notebook(headers, workspace_id, name, notebook_content):
    """Deploy a notebook with its full definition (content)."""
    encoded = base64.b64encode(notebook_content.encode()).decode()
    payload = {
        "displayName": name,
        "type": "Notebook",
        "definition": {
            "format": "ipynb",
            "parts": [
                {
                    "path": "notebook-content.py",
                    "payload": encoded,
                    "payloadType": "InlineBase64"
                }
            ]
        }
    }

    response = requests.post(
        f"https://api.fabric.microsoft.com/v1/workspaces/{workspace_id}/items",
        headers=headers,
        json=payload
    )
    # 202 = long-running; poll for completion
    if response.status_code == 202:
        return poll_operation(headers, response)
    response.raise_for_status()
    return response.json()

import time

def poll_operation(headers, initial_response, timeout=300, interval=5):
    """Poll a long-running operation until completion."""
    operation_url = initial_response.headers.get("Location")
    if not operation_url:
        # Retry-After header may also be present
        operation_id = initial_response.headers.get("x-ms-operation-id")
        operation_url = f"https://api.fabric.microsoft.com/v1/operations/{operation_id}"

    elapsed = 0
    while elapsed < timeout:
        resp = requests.get(operation_url, headers=headers)
        resp.raise_for_status()
        status = resp.json().get("status")

        if status == "Succeeded":
            result_url = resp.json().get("resourceLocation")
            if result_url:
                return requests.get(result_url, headers=headers).json()
            return resp.json()
        elif status in ("Failed", "Cancelled"):
            raise Exception(f"Operation {status}: {resp.json()}")

        retry_after = int(resp.headers.get("Retry-After", interval))
        time.sleep(retry_after)
        elapsed += retry_after

    raise TimeoutError(f"Operation did not complete within {timeout}s")

def scale_capacity(subscription_id, resource_group, capacity_name, sku):
    """Scale a Fabric capacity to a different SKU."""
    from azure.identity import DefaultAzureCredential
    credential = DefaultAzureCredential()
    token = credential.get_token("https://management.azure.com/.default")

    url = (
        f"https://management.azure.com/subscriptions/{subscription_id}"
        f"/resourceGroups/{resource_group}"
        f"/providers/Microsoft.Fabric/capacities/{capacity_name}"
        f"?api-version=2023-11-01"
    )

    payload = {
        "sku": {"name": sku, "tier": "Fabric"},  # F2, F4, F8, ..., F2048
        "properties": {"administration": {"members": ["admin@contoso.com"]}}
    }

    response = requests.patch(
        url,
        headers={"Authorization": f"Bearer {token.token}", "Content-Type": "application/json"},
        json=payload
    )
    response.raise_for_status()
    return response.json()

def pause_capacity(subscription_id, rg, name):
    """Pause (suspend) a Fabric capacity to stop billing."""
    url = (f"https://management.azure.com/subscriptions/{subscription_id}"
           f"/resourceGroups/{rg}/providers/Microsoft.Fabric/capacities/{name}"
           f"/suspend?api-version=2023-11-01")
    # POST with empty body
    response = requests.post(url, headers=arm_headers)
    return poll_arm_operation(response)

def resume_capacity(subscription_id, rg, name):
    """Resume a paused Fabric capacity."""
    url = (f"https://management.azure.com/subscriptions/{subscription_id}"
           f"/resourceGroups/{rg}/providers/Microsoft.Fabric/capacities/{name}"
           f"/resume?api-version=2023-11-01")
    response = requests.post(url, headers=arm_headers)
    return poll_arm_operation(response)

def get_all_pages(headers, url):
    """Fetch all pages of a paginated Fabric API response."""
    all_items = []
    while url:
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        data = response.json()
        all_items.extend(data.get("value", []))
        url = data.get("continuationUri")  # None when no more pages
    return all_items

def api_call_with_retry(method, url, headers, json=None, max_retries=5):
    """Make an API call with exponential backoff for transient errors."""
    for attempt in range(max_retries):
        response = requests.request(method, url, headers=headers, json=json)
        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 2 ** attempt))
            time.sleep(retry_after)
            continue
        if response.status_code >= 500:
            time.sleep(2 ** attempt)
            continue
        return response
    raise Exception(f"API call failed after {max_retries} retries")

CASINO_PROPERTIES = [
    {"name": "Bellagio", "region": "westus2", "code": "BEL"},
    {"name": "MGM Grand", "region": "westus2", "code": "MGM"},
    {"name": "Borgata", "region": "eastus2", "code": "BOR"},
]

def provision_casino_workspaces(headers, capacity_id):
    """Provision standardized workspaces for each casino property."""
    for prop in CASINO_PROPERTIES:
        for env in ["dev", "staging", "prod"]:
            ws_name = f"casino-{prop['code'].lower()}-{env}"
            ws = create_workspace(headers, ws_name, capacity_id,
                                  description=f"{prop['name']} - {env.upper()}")
            # Create standard lakehouses
            for layer in ["lh_bronze", "lh_silver", "lh_gold"]:
                create_lakehouse(headers, ws["id"], f"{layer}_{prop['code'].lower()}")
            print(f"Provisioned {ws_name} with medallion lakehouses")

import glob

def deploy_all_notebooks(headers, workspace_id, notebook_dir):
    """Deploy all notebooks from a local directory to a Fabric workspace."""
    notebooks = glob.glob(f"{notebook_dir}/**/*.py", recursive=True)
    results = {"success": [], "failed": []}

    for nb_path in sorted(notebooks):
        name = nb_path.split("/")[-1].replace(".py", "")
        with open(nb_path, "r") as f:
            content = f.read()
        try:
            deploy_notebook(headers, workspace_id, name, content)
            results["success"].append(name)
        except Exception as e:
            results["failed"].append({"name": name, "error": str(e)})

    return results

# Deploy bronze, silver, gold notebooks
deploy_all_notebooks(headers, prod_ws_id, "notebooks/bronze")
deploy_all_notebooks(headers, prod_ws_id, "notebooks/silver")
deploy_all_notebooks(headers, prod_ws_id, "notebooks/gold")

def generate_compliance_report(headers):
    """Scan all workspaces and generate a compliance posture report."""
    # Get all workspaces in the tenant
    workspaces = get_all_pages(
        headers,
        "https://api.fabric.microsoft.com/v1/admin/workspaces?$top=100"
    )

    report = []
    for ws in workspaces:
        ws_detail = requests.get(
            f"https://api.fabric.microsoft.com/v1/admin/workspaces/{ws['id']}",
            headers=headers
        ).json()

        items = get_all_pages(
            headers,
            f"https://api.fabric.microsoft.com/v1/admin/workspaces/{ws['id']}/items"
        )

        report.append({
            "workspace": ws["displayName"],
            "capacity": ws.get("capacityId"),
            "state": ws.get("state"),
            "item_count": len(items),
            "item_types": list(set(i["type"] for i in items)),
            "has_sensitivity_label": ws.get("sensitivityLabel") is not None,
            "role_assignments": len(ws_detail.get("roleAssignments", []))
        })

    return report

def provision_usda_grant_workspaces(headers, capacity_id, fiscal_year):
    """Create workspaces per USDA grant program with compliance controls."""
    grant_programs = ["SNAP", "WIC", "CRP", "EQIP", "FSA_Loans"]

    for program in grant_programs:
        ws_name = f"usda-{program.lower()}-fy{fiscal_year}"
        ws = create_workspace(headers, ws_name, capacity_id,
                              description=f"USDA {program} - FY{fiscal_year}")

        # Apply sensitivity label via Admin API
        requests.post(
            f"https://api.fabric.microsoft.com/v1/admin/workspaces/{ws['id']}/assignLabel",
            headers=headers,
            json={"labelId": "federal-cui-label-id"}
        )

        # Create standard lakehouses
        for lh in ["lh_bronze", "lh_silver", "lh_gold"]:
            create_lakehouse(headers, ws["id"], f"{lh}_{program.lower()}")