Synapse Migration

import re
from urllib.parse import unquote

def parse_synapse_url(url):
    """Extract subscription_id, resource_group, workspace_name from a Synapse Studio URL."""
    decoded = unquote(url)
    m = re.search(
        r'/subscriptions/([^/]+)/resourceGroups/([^/]+)/providers/Microsoft\.Synapse/workspaces/([^/?\s&]+)',
        decoded, re.IGNORECASE
    )
    if m:
        return {"subscription_id": m.group(1), "resource_group": m.group(2), "workspace_name": m.group(3)}
    return None

Migration Orchestrator:
│
├── Step 0: Strategy decision (lift-and-shift vs. migrate-and-modernize)
├── Step 1: Authenticate (acquire 3 tokens)
├── Step 2: Full inventory (Spark pools, databases, notebooks, SJDs)
│
├── Phase 0: Spark Pools → Environments
│   ├── Execute spark-pool-migration.md Steps 1–6
│   ├── ⏸ GATE: Validate Environments (V1)
│   └── Output: poolMappings JSON
│
├── Phase 1: Lake Databases / HMS → Lakehouses
│   ├── Detect: built-in vs. external HMS
│   ├── Execute lake-database-migration.md or external-hms-migration.md
│   ├── ⏸ GATE: Validate data (V2)
│   └── Output: lakehouseMappings JSON
│
├── Phase 1b: Storage Path Inventory (migrate-and-modernize only)
│   ├── Scan notebook/SJD code for abfss:// paths
│   ├── Deduplicate by storage account + container
│   ├── Create OneLake shortcuts for each unique container
│   └── Output: shortcutMappings JSON (consumed by Phase 2 path rewriting)
│
├── Phase 2: Notebooks → Fabric Notebooks
│   ├── Execute spark-item-migration.md Phase 2 Steps 1–6
│   ├── ⏸ GATE: Validate notebooks (V3)
│   └── Output: notebookMappings JSON
│
├── Phase 3: SJDs → Fabric SJDs
│   ├── Execute spark-item-migration.md Phase 3 Steps 1–4
│   ├── ⏸ GATE: Validate SJDs (V4)
│   └── Output: sjdMappings JSON
│
├── Final Validation: V5 (query comparison) + V6 (full report)
│
├── Migration Report: Query Fabric workspace → generate Markdown
│   ├── Per-phase tables with clickable portal links
│   ├── Blockers & post-migration actions
│   └── Output: migration-report-{migrationId}.md
│
└── (Optional) Security & Governance — for production deployment
    └── Apply when promoting validated migration to production

import subprocess, json, requests, time, shutil

def get_token(resource):
    """Acquire a token via az CLI."""
    az_path = shutil.which("az")
    if not az_path:
        raise RuntimeError(f"az CLI not found in PATH")
    result = subprocess.run(
        [az_path, "account", "get-access-token", "--resource", resource,
         "--query", "accessToken", "-o", "tsv"],
        capture_output=True, text=True, check=True
    )
    return result.stdout.strip()

class TokenManager:
    """Auto-refreshing token manager for multi-phase migrations.
    Tokens expire after ~60 min. For large workspaces (50+ items),
    migration routinely exceeds this. The manager detects 401 responses
    and transparently re-acquires all three tokens."""

    def __init__(self):
        self.refresh_all()

    def refresh_all(self):
        self.arm = get_token("https://management.azure.com")
        self.synapse = get_token("https://dev.azuresynapse.net")
        self.fabric = get_token("https://api.fabric.microsoft.com")
        self._update_headers()

    def _update_headers(self):
        self.arm_headers = {"Authorization": f"Bearer {self.arm}"}
        self.syn_headers = {"Authorization": f"Bearer {self.synapse}"}
        self.fab_headers = {
            "Authorization": f"Bearer {self.fabric}",
            "Content-Type": "application/json"
        }

    def refresh_if_needed(self, resp):
        """If response is 401 TokenExpired, refresh all tokens and return True."""
        if resp.status_code == 401:
            self.refresh_all()
            return True
        return False

tokens = TokenManager()

# Verify Synapse ARM access
resp = requests.get(
    f"https://management.azure.com/subscriptions/{sub_id}/resourceGroups/{rg}"
    f"/providers/Microsoft.Synapse/workspaces/{ws_name}?api-version=2021-06-01",
    headers=tokens.arm_headers
)
assert resp.status_code == 200, f"ARM access failed: {resp.status_code} {resp.text}"

# Verify Synapse data-plane access
resp = requests.get(
    f"https://{ws_name}.dev.azuresynapse.net/notebooks?api-version=2020-12-01",
    headers=tokens.syn_headers
)
assert resp.status_code == 200, f"Synapse data-plane access failed: {resp.status_code}"

# Resolve Fabric workspace ID
resp = requests.get(
    "https://api.fabric.microsoft.com/v1/workspaces",
    headers=tokens.fab_headers
)
workspaces = resp.json()["value"]
fabric_ws = next((w for w in workspaces if w["displayName"] == fabric_ws_name), None)
assert fabric_ws, f"Fabric workspace '{fabric_ws_name}' not found"
fabric_ws_id = fabric_ws["id"]

def existing_items_map(item_type):
    """Get map of displayName → id for existing items. Used to skip duplicates on resume."""
    resp = requests.get(
        f"https://api.fabric.microsoft.com/v1/workspaces/{fabric_ws_id}/items?type={item_type}",
        headers=tokens.fab_headers
    )
    if tokens.refresh_if_needed(resp):
        resp = requests.get(
            f"https://api.fabric.microsoft.com/v1/workspaces/{fabric_ws_id}/items?type={item_type}",
            headers=tokens.fab_headers
        )
    result = {}
    if resp.status_code == 200:
        for item in resp.json().get("value", []):
            result[item["displayName"]] = item["id"]
    return result

# Phase 0 example
existing_envs = existing_items_map("Environment")
for pool in pools:
    env_name = f"{pool_name}_env"
    if env_name in existing_envs:
        env_id = existing_envs[env_name]
        # Record in mappings and skip creation
        continue
    # ... create new Environment

pools_resp = requests.get(
    f"https://management.azure.com/subscriptions/{sub_id}/resourceGroups/{rg}"
    f"/providers/Microsoft.Synapse/workspaces/{ws_name}/bigDataPools?api-version=2021-06-01",
    headers=tokens.arm_headers
)
pools = pools_resp.json()["value"]

external_hms_pools = []
for pool in pools:
    config_content = pool.get("properties", {}).get("sparkConfigProperties", {}).get("content", "")
    if "javax.jdo.option.ConnectionURL" in config_content:
        external_hms_pools.append(pool["name"])
has_external_hms = len(external_hms_pools) > 0

dbs_resp = requests.get(
    f"https://{ws_name}.dev.azuresynapse.net/databases?api-version=2021-04-01",
    headers=tokens.syn_headers
)
databases = [db for db in dbs_resp.json().get("value", [])
             if db.get("properties", {}).get("origin", {}).get("type") != "SQLOD"]

all_tables = {}
for db in databases:
    db_name = db["name"]
    tables_resp = requests.get(
        f"https://{ws_name}.dev.azuresynapse.net/databases/{db_name}/TABLEs?api-version=2021-04-01",
        headers=tokens.syn_headers
    )
    all_tables[db_name] = tables_resp.json().get("value", [])

notebooks_resp = requests.get(
    f"https://{ws_name}.dev.azuresynapse.net/notebooks?api-version=2020-12-01",
    headers=tokens.syn_headers
)
notebooks = notebooks_resp.json()["value"]
# Handle pagination
while "nextLink" in notebooks_resp.json():
    notebooks_resp = requests.get(notebooks_resp.json()["nextLink"],
                                  headers=tokens.syn_headers)
    notebooks.extend(notebooks_resp.json()["value"])

sjds_resp = requests.get(
    f"https://{ws_name}.dev.azuresynapse.net/sparkJobDefinitions?api-version=2020-12-01",
    headers=tokens.syn_headers
)
sjds = sjds_resp.json()["value"]

gpu_count = sum(
    1 for p in pools
    if "GPU" in p.get("properties", {}).get("nodeSizeFamily", "")
)
delta_count = sum(
    1 for tables in all_tables.values() for t in tables
    if t.get("properties", {}).get("storageDescriptor", {}).get("format", {}).get("formatType", "").lower() == "delta"
)
non_delta_count = sum(len(t) for t in all_tables.values()) - delta_count

╔══════════════════════════════════════════════════════════════════╗
║                    SYNAPSE WORKSPACE INVENTORY                  ║
║  Source: {ws_name}                                              ║
║  Target: {fabric_ws_name} ({fabric_ws_id})                     ║
╠══════════════════════════════════════════════════════════════════╣
║                                                                  ║
║  Spark Pools:        {len(pools)}                               ║
║    - GPU pools:      {gpu_count} (MIGRATION BLOCKER if > 0)    ║
║    - External HMS:   {len(external_hms_pools)} pools            ║
║                                                                  ║
║  Lake Databases:     {len(databases)}                           ║
║    - Total tables:   {sum(len(t) for t in all_tables.values())} ║
║    - Delta tables:   {delta_count}                              ║
║    - Non-Delta:      {non_delta_count}                          ║
║                                                                  ║
║  Notebooks:          {len(notebooks)}                           ║
║  Spark Job Defs:     {len(sjds)}                                ║
║                                                                  ║
║  Strategy:           {strategy}                                 ║
║  HMS Type:           {"External" if has_external_hms else "Built-in"} ║
╚══════════════════════════════════════════════════════════════════╝

import time, requests

def update_env_definition(workspace_id, env_id, definition_payload, max_retries=3):
    """Upload Environment definition with retry for post-creation race condition
    and automatic token refresh on 401.
    After POST /environments → 201, the updateDefinition endpoint may return 404
    for a few seconds while the Environment is provisioned."""
    url = (f"https://api.fabric.microsoft.com/v1/workspaces/{workspace_id}"
           f"/environments/{env_id}/updateDefinition")
    for attempt in range(1, max_retries + 1):
        resp = requests.post(url, headers=tokens.fab_headers, json=definition_payload)
        if resp.status_code == 401:
            tokens.refresh_if_needed(resp)
            resp = requests.post(url, headers=tokens.fab_headers, json=definition_payload)
            return resp
        if resp.status_code == 404 and attempt < max_retries:
            time.sleep(3 * attempt)  # 3s, 6s backoff
            continue
        resp.raise_for_status()
        return resp
    resp.raise_for_status()

# Validate every Environment
for pool_name, mapping in pool_mappings.items():
    if mapping["mode"] == "starterPool":
        continue
    env_id = mapping["fabricEnvironmentId"]
    resp = requests.get(
        f"https://api.fabric.microsoft.com/v1/workspaces/{fabric_ws_id}/environments/{env_id}",
        headers=tokens.fab_headers
    )
    publish_state = resp.json()["properties"]["publishDetails"]["state"]
    assert publish_state == "Success", f"Environment {pool_name}_env publish failed: {publish_state}"

{
  "poolMappings": {
    "{poolName}": {
      "fabricEnvironmentId": "{envId}",
      "fabricEnvironmentName": "{poolName}_env",
      "mode": "customPool | librariesOnly | starterPool"
    }
  }
}

Has external HMS? (detected in Step 2b)
├── YES → Follow external-hms-migration.md Steps 0–8
│         (JDBC connection details needed — ask user for credentials, even in lift-and-shift)
└── NO  → Follow lake-database-migration.md Steps 1–8

Databases found: sales, marketing, staging, default

Choose mapping mode:
  A) Schemas — all databases → schemas in one Lakehouse (simplest)
  B) Separate — each database → its own Lakehouse (most isolation)
  C) Hybrid — assign each database individually (most flexible)

# Validate shortcuts
for db_name, tables in all_tables.items():
    schema = lakehouse_mappings[db_name]["schema"]
    for table in tables:
        table_name = table["name"]
        format_type = table["properties"]["storageDescriptor"]["format"]["formatType"]
        section = "Tables" if format_type == "delta" else "Files"
        # List shortcut contents via Fabric API or notebook
        # Verify shortcut resolves without error

{
  "lakehouseMappings": {
    "{databaseName}": {
      "lakehouseId": "{lhId}",
      "lakehouseName": "{lhName}",
      "schema": "{fabricSchemaName}",
      "mode": "dedicated | schema"
    }
  }
}

import re

abfss_pattern = re.compile(
    r'abfss://([^@]+)@([^.]+)\.dfs\.core\.windows\.net/([^\s\'")\]]*)'
)

storage_refs = {}  # key: (container, account) → set of subpaths

for nb in notebooks:
    for cell in nb["properties"]["cells"]:
        if cell["cell_type"] != "code":
            continue
        source = "".join(cell["source"])
        for match in abfss_pattern.finditer(source):
            container, account, subpath = match.groups()
            key = (container, account)
            # Extract the top-level directory as the shortcut target
            top_dir = subpath.split("/")[0] if subpath else ""
            storage_refs.setdefault(key, set()).add(top_dir)

for sjd in sjds_source:
    file_path = sjd["properties"].get("jobProperties", {}).get("file", "")
    for match in abfss_pattern.finditer(file_path):
        container, account, subpath = match.groups()
        key = (container, account)
        top_dir = subpath.split("/")[0] if subpath else ""
        storage_refs.setdefault(key, set()).add(top_dir)

# phase1_shortcuts: set of (container, account) already covered by Phase 1
phase1_shortcuts = set()
for db_name, mapping in lakehouse_mappings.items():
    for table in mapping.get("tables", []):
        loc = table.get("sourceLocation", "")
        m = abfss_pattern.match(loc)
        if m:
            phase1_shortcuts.add((m.group(1), m.group(2)))

# Only create shortcuts for paths NOT already covered
new_shortcuts = {k: v for k, v in storage_refs.items() if k not in phase1_shortcuts}

lakehouse_id = list(lakehouse_mappings.values())[0]["lakehouseId"]  # primary Lakehouse

shortcut_mappings = {}

for (container, account), subdirs in new_shortcuts.items():
    shortcut_name = f"{account}_{container}"  # e.g., "mystorageaccount_raw"
    payload = {
        "name": shortcut_name,
        "path": "Files",
        "target": {
            "type": "AdlsGen2",
            "adlsGen2": {
                "location": f"https://{account}.dfs.core.windows.net",
                "subpath": f"/{container}",
                "connectionId": adls_connection_id
            }
        }
    }
    resp = requests.post(
        f"{BASE_URL}/workspaces/{fabric_ws_id}/items/{lakehouse_id}/shortcuts",
        headers=tokens.fab_headers,
        json=payload
    )
    if resp.status_code == 409:
        # Shortcut already exists — reuse it
        pass
    else:
        resp.raise_for_status()

    # Build the path mapping: original abfss:// → OneLake relative path
    shortcut_mappings[(container, account)] = {
        "shortcutName": shortcut_name,
        "lakehouseId": lakehouse_id,
        "onelakePath": f"Files/{shortcut_name}",
        # Notebooks can use relative path: Files/{shortcut_name}/{subpath}
    }

# Convert to serializable format
shortcut_mappings_json = {}
for (container, account), mapping in shortcut_mappings.items():
    key = f"abfss://{container}@{account}.dfs.core.windows.net"
    shortcut_mappings_json[key] = mapping

{
  "shortcutMappings": {
    "abfss://[email protected]": {
      "shortcutName": "mystorageaccount_raw",
      "lakehouseId": "{lhId}",
      "onelakePath": "Files/mystorageaccount_raw"
    },
    "abfss://[email protected]": {
      "shortcutName": "mystorageaccount_processed",
      "lakehouseId": "{lhId}",
      "onelakePath": "Files/mystorageaccount_processed"
    }
  }
}

search_patterns = [
    "mssparkutils", "spark.synapse.linkedService", "getSecretWithLS",
    "TokenLibrary", "synapsesql", "spark.catalog.listDatabases",
    "spark.catalog.currentDatabase", "spark.catalog.getDatabase",
    "LinkedServiceBasedTokenProvider", "getPropertiesAsMap",
    "spark.storage.synapse", "/user/trusted-service-user/",
    "cosmos.oltp", "kusto.spark.synapse"
]

audit = {}
for nb in notebooks:
    nb_name = nb["name"]
    hits = []
    for cell in nb["properties"]["cells"]:
        if cell["cell_type"] != "code":
            continue
        source = "".join(cell["source"])
        for pattern in search_patterns:
            if pattern in source:
                hits.append(pattern)
    if hits:
        audit[nb_name] = list(set(hits))
    # else: safe to lift-and-shift

Notebook Code Audit:
  Clean (no Synapse-specific code):  85 notebooks
  Has Synapse-specific patterns:     37 notebooks

  Top patterns found:
    mssparkutils           → 35 notebooks  (replace with notebookutils)
    TokenLibrary           → 12 notebooks  (replace with direct OAuth)
    synapsesql             →  8 notebooks  (replace with Delta read)
    cosmos.oltp            →  3 notebooks  (update auth)
    kusto.spark.synapse    →  2 notebooks  (update auth)

warnings = []

FLAG_ANCHORS = {
    "SYNAPSESQL_NO_EQUIVALENT": "g1-sparkreadsynapsesql--no-direct-fabric-equivalent",
    "LIBRARY_VERSION_CONFLICT": "g2-custom-library-version-conflicts-with-fabric-runtime",
    "DELTA_PROTOCOL_MISMATCH": "g3-delta-lake-protocol-version-incompatibility",
    "SECURITY_MODEL_INCOMPATIBLE": "g4-synapse-security-model--managed-identities--ip-firewall",
    "GPU_POOL_UNSUPPORTED": "g5-gpu-pool-migration-blocker",
    "DOTNET_SPARK_UNSUPPORTED": "g6-net-for-spark-cf-sjd-blocker",
    "SESSION_CONFIG_IGNORED": "g7-notebook-session-configuration-configure",
    "SHORTCUT_CONNECTION_FAILED": "g8-shortcut-creation-failures-adls-connection-issues",
}

def flag_warning(item_name, flag_id, severity, details):
    warnings.append({
        "item": item_name, "flag": flag_id,
        "severity": severity, "details": details,
        "anchor": FLAG_ANCHORS.get(flag_id, flag_id.lower()),
    })

# Phase 0 — GPU pool blockers
for pool in pools:
    if pool["properties"].get("nodeSizeFamily") == "HardwareAcceleratedGPU":
        flag_warning(pool["name"], "GPU_POOL_UNSUPPORTED", "High",
                     "GPU pool — no Fabric equivalent")

# Phase 2 — code audit hits
PATTERN_FLAGS = {
    "synapsesql":       ("SYNAPSESQL_NO_EQUIVALENT", "High",
                         "Uses spark.read.synapsesql() — replace with JDBC or shortcut"),
    "LinkedServiceBasedTokenProvider": ("SECURITY_MODEL_INCOMPATIBLE", "Medium",
                         "Uses LinkedServiceBasedTokenProvider — replace with ClientCredsTokenProvider"),
    "getPropertiesAsMap": ("SECURITY_MODEL_INCOMPATIBLE", "Medium",
                         "Uses TokenLibrary.getPropertiesAsMap() — no Fabric equivalent"),
}

for nb_name, patterns in audit.items():
    for pat in patterns:
        if pat in PATTERN_FLAGS:
            fid, sev, det = PATTERN_FLAGS[pat]
            flag_warning(nb_name, fid, sev, det)

# Phase 3 — .NET SJD blockers
for sjd in sjds_source:
    lang = sjd["properties"].get("language", "")
    if lang.lower() in ("dotnet", "csharp"):
        flag_warning(sjd["name"], "DOTNET_SPARK_UNSUPPORTED", "High",
                     f".NET SJD (language={lang}) — rewrite in Python or Scala")

# Run each notebook via Job API
for nb_name, mapping in notebook_mappings.items():
    nb_id = mapping["fabricNotebookId"]
    resp = requests.post(
        f"https://api.fabric.microsoft.com/v1/workspaces/{fabric_ws_id}"
        f"/items/{nb_id}/jobs/instances?jobType=RunNotebook",
        headers=tokens.fab_headers,
        json={}
    )
    # Poll for completion...

{
  "notebookMappings": {
    "{notebookName}": {
      "synapseNotebookName": "{originalName}",
      "fabricNotebookId": "{nbId}",
      "refactored": true,
      "auditPatterns": ["mssparkutils", "TokenLibrary"],
      "boundLakehouse": "{lakehouseId}",
      "boundEnvironment": "{environmentId}"
    }
  }
}

# Run each SJD via Job API
for sjd_name, mapping in sjd_mappings.items():
    sjd_id = mapping["fabricSjdId"]
    resp = requests.post(
        f"https://api.fabric.microsoft.com/v1/workspaces/{fabric_ws_id}"
        f"/items/{sjd_id}/jobs/instances?jobType=SparkJob",
        headers=tokens.fab_headers,
        json={}
    )
    # Poll for completion...

{
  "sjdMappings": {
    "{sjdName}": {
      "synapseSjdName": "{originalName}",
      "fabricSjdId": "{sjdId}",
      "boundEnvironment": "{environmentId}",
      "language": "python | scala"
    }
  }
}

# Collect all mappings into a single migration state
migration_state = {
    "source": {"workspace": ws_name, "subscription": sub_id, "resourceGroup": rg},
    "target": {"workspace": fabric_ws_name, "workspaceId": fabric_ws_id},
    "strategy": strategy,
    "poolMappings": pool_mappings,
    "lakehouseMappings": lakehouse_mappings,
    "notebookMappings": notebook_mappings,
    "sjdMappings": sjd_mappings,
}

╔══════════════════════════════════════════════════════════════════╗
║                    MIGRATION VALIDATION REPORT                  ║
╠══════════════════════════════════════════════════════════════════╣
║  Environments:  {env_pass}/{env_total} passed                   ║
║  Lakehouses:    {lh_pass}/{lh_total} passed                     ║
║  Shortcuts:     {sc_pass}/{sc_total} passed                     ║
║  Row counts:    {rc_pass}/{rc_total} matched                    ║
║  Notebooks:     {nb_pass}/{nb_total} executed successfully      ║
║  SJDs:          {sjd_pass}/{sjd_total} executed successfully    ║
║  Checksums:     {cs_pass}/{cs_total} matched                    ║
╠══════════════════════════════════════════════════════════════════╣
║  VERDICT:       {READY FOR CUTOVER | ISSUES REMAIN}            ║
╚══════════════════════════════════════════════════════════════════╝

import requests, json, base64, os
from datetime import datetime, timezone

# --- Inputs (from migration state) ---
FABRIC_TOKEN = "<fabric-token>"
WORKSPACE_ID = migration_state["target"]["workspaceId"]
SYNAPSE_WS = migration_state["source"]["workspaceName"]
SUB_ID = migration_state["source"]["subscriptionId"]
RG = migration_state["source"]["resourceGroup"]
STRATEGY = migration_state["strategy"]
shortcut_mappings = migration_state.get("shortcutMappings", {})  # populated by Phase 1b; empty for lift-and-shift

PORTAL_URL = "https://app.fabric.microsoft.com"
FAB_BASE = "https://api.fabric.microsoft.com/v1"
headers = {"Authorization": f"Bearer {FABRIC_TOKEN}"}

# --- Collect all items from Fabric workspace ---
def list_items(item_type=None):
    url = f"{FAB_BASE}/workspaces/{WORKSPACE_ID}/items"
    if item_type:
        url += f"?type={item_type}"
    items = []
    while url:
        resp = requests.get(url, headers=headers)
        resp.raise_for_status()
        data = resp.json()
        items.extend(data.get("value", []))
        url = data.get("continuationUri")
    return items

environments = list_items("Environment")
lakehouses = list_items("Lakehouse")
notebooks = list_items("Notebook")
sjds = list_items("SparkJobDefinition")

# --- URL builder ---
def item_url(item):
    itype = item["type"]
    iid = item["id"]
    type_paths = {
        "Lakehouse": f"lakehouses/{iid}",
        "Notebook": f"synapse/notebooks/{iid}",
        "SparkJobDefinition": f"synapse/sparkjobdefinitions/{iid}",
        "Environment": f"environments/{iid}",
    }
    path = type_paths.get(itype, f"items/{iid}")
    return f"{PORTAL_URL}/groups/{WORKSPACE_ID}/{path}"

# --- Build report ---
lines = []
lines.append("# Synapse → Fabric Migration Report")
lines.append("")
lines.append(f"**Generated**: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}")
lines.append(f"**Source**: Synapse workspace `{SYNAPSE_WS}` (sub: `{SUB_ID}`, rg: `{RG}`)")
lines.append(f"**Target**: [Fabric Workspace]({PORTAL_URL}/groups/{WORKSPACE_ID})")
lines.append(f"**Strategy**: {STRATEGY}")
lines.append("")

# Phase 0: Environments
lines.append("## Phase 0: Spark Pool → Environment")
lines.append("")
lines.append("| # | Environment | Source Pool | Link | Status |")
lines.append("|---|---|---|---|---|")
for i, env in enumerate(environments, 1):
    link = item_url(env)
    desc = env.get("description", "")
    pool = desc.replace("Migrated from Synapse Spark pool: ", "") if "Migrated from" in desc else "—"
    lines.append(f"| {i} | `{env['displayName']}` | {pool} | [Open]({link}) | ✓ |")
lines.append("")

# Phase 1: Lakehouses
lines.append("## Phase 1: Lake Database → Lakehouse")
lines.append("")
if lakehouses:
    lines.append("| # | Lakehouse | Source DB | Link | Shortcuts | Status |")
    lines.append("|---|---|---|---|---|---|")
    for i, lh in enumerate(lakehouses, 1):
        link = item_url(lh)
        desc = lh.get("description", "")
        db = desc.replace("Migrated from Synapse database: ", "") if "Migrated from" in desc else "—"
        # Shortcut count from migration state if available
        sc = migration_state.get("lakehouseMappings", {}).get(lh["displayName"], {}).get("shortcutCount", "—")
        lines.append(f"| {i} | `{lh['displayName']}` | {db} | [Open]({link}) | {sc} | ✓ |")
else:
    lines.append("No Lakehouses created (source had 0 Lake Databases or Phase 1 was skipped).")
lines.append("")

# Phase 1b: Storage Path Shortcuts (migrate-and-modernize only)
if shortcut_mappings:  # initialized in Phase 1b Step 3; empty dict in lift-and-shift mode
    lines.append("## Phase 1b: Storage Path Shortcuts")
    lines.append("")
    lines.append("OneLake shortcuts created for ad-hoc `abfss://` storage paths found in notebook/SJD code.")
    lines.append("")
    lines.append("| # | Shortcut Name | Source Storage | Container | Lakehouse | Path Refs |")
    lines.append("|---|---|---|---|---|---|")
    for i, (prefix, mapping) in enumerate(shortcut_mappings.items(), 1):
        # Parse account and container from prefix
        parts = prefix.replace("abfss://", "").split("@")
        container = parts[0] if parts else "—"
        account = parts[1].replace(".dfs.core.windows.net", "") if len(parts) > 1 else "—"
        lh_name = next(
            (lh["displayName"] for lh in lakehouses if lh["id"] == mapping["lakehouseId"]),
            mapping["lakehouseId"]
        )
        lines.append(
            f"| {i} | `{mapping['shortcutName']}` | `{account}` | `{container}` "
            f"| `{lh_name}` | `{mapping['onelakePath']}` |"
        )
    lines.append("")
    lines.append(f"**Total shortcuts**: {len(shortcut_mappings)}")
    lines.append("")

# Phase 2: Notebooks
lines.append("## Phase 2: Notebooks")
lines.append("")
lines.append("| # | Notebook | Source | Link | Flagged Patterns | Status |")
lines.append("|---|---|---|---|---|---|")
for i, nb in enumerate(notebooks, 1):
    link = item_url(nb)
    desc = nb.get("description", "")
    source = desc.replace("Migrated from Synapse: ", "") if "Migrated from" in desc else "—"
    # Check migration state for flagged patterns
    nb_state = migration_state.get("notebookMappings", {}).get(nb["displayName"], {})
    flags = nb_state.get("flaggedPatterns", [])
    flag_str = ", ".join(f"`{f}`" for f in flags) if flags else "—"
    status = "⚠ Needs refactoring" if flags else "✓"
    lines.append(f"| {i} | `{nb['displayName']}` | {source} | [Open]({link}) | {flag_str} | {status} |")
lines.append("")

# Phase 3: SJDs
lines.append("## Phase 3: Spark Job Definitions")
lines.append("")
lines.append("| # | SJD | Source | Language | Link | Status |")
lines.append("|---|---|---|---|---|---|")
for i, sjd in enumerate(sjds, 1):
    link = item_url(sjd)
    desc = sjd.get("description", "")
    source, lang = "—", "—"
    if "Migrated from Synapse SJD: " in desc:
        rest = desc.replace("Migrated from Synapse SJD: ", "")
        if " (language: " in rest:
            source, lang_part = rest.rsplit(" (language: ", 1)
            lang = lang_part.rstrip(")")
        else:
            source = rest
    lines.append(f"| {i} | `{sjd['displayName']}` | {source} | {lang} | [Open]({link}) | ✓ |")
lines.append("")

# Blockers (from migration state errors)
blockers = [e for e in migration_state.get("errors", []) if e.get("severity") == "blocker"]
if blockers:
    lines.append("## Migration Blockers")
    lines.append("")
    lines.append("| # | Source Item | Phase | Reason |")
    lines.append("|---|---|---|---|")
    for i, b in enumerate(blockers, 1):
        lines.append(f"| {i} | `{b['item']}` | {b['phase']} | {b['error']} |")
    lines.append("")

# Post-migration actions
lines.append("## Post-Migration Actions")
lines.append("")
lines.append("| Priority | Action | Items Affected |")
lines.append("|---|---|---|")
# Auto-generate actions from flagged patterns
refactor_nbs = [nb["displayName"] for nb in notebooks
                if migration_state.get("notebookMappings", {}).get(nb["displayName"], {}).get("flaggedPatterns")]
if refactor_nbs:
    lines.append(f"| **High** | Refactor Synapse-specific patterns (`mssparkutils`, `TokenLibrary`, etc.) | {', '.join(f'`{n}`' for n in refactor_nbs)} |")
if blockers:
    lines.append(f"| **Medium** | Resolve migration blockers (see table above) | {len(blockers)} item(s) |")
lines.append("| **Low** | Run security & governance setup for production | See [security-governance.md](security-governance.md) |")
lines.append("")

# Summary
total = len(environments) + len(lakehouses) + len(notebooks) + len(sjds)
lines.append("## Summary")
lines.append("")
lines.append("| Phase | Type | Count |")
lines.append("|---|---|---|")
lines.append(f"| 0 | Environments | {len(environments)} |")
lines.append(f"| 1 | Lakehouses | {len(lakehouses)} |")
lines.append(f"| 2 | Notebooks | {len(notebooks)} |")
lines.append(f"| 3 | Spark Job Definitions | {len(sjds)} |")
lines.append(f"| **Total** | | **{total}** |")

report = "\n".join(lines)
print(report)

# Save report
report_path = f"migration-report-{migration_state.get('migrationId', 'output')}.md"
with open(report_path, "w", encoding="utf-8") as f:
    f.write(report)
print(f"\nReport saved to {report_path}")

migration_state["reportGeneratedAt"] = datetime.now(timezone.utc).isoformat()
migration_state["reportPath"] = report_path
migration_state["summary"] = {
    "environments": len(environments),
    "lakehouses": len(lakehouses),
    "notebooks": len(notebooks),
    "sjds": len(sjds),
    "total": total,
    "blockers": len(blockers),
}
# Save updated state
with open(state_file_path, "w") as f:
    json.dump(migration_state, f, indent=2)

{
  "migrationId": "{uuid}",
  "startedAt": "2026-04-24T10:00:00Z",
  "strategy": "migrate-and-modernize",
  "source": {
    "workspaceName": "my-synapse-ws",
    "subscriptionId": "...",
    "resourceGroup": "my-rg"
  },
  "target": {
    "workspaceName": "Migration-Target",
    "workspaceId": "..."
  },
  "phases": {
    "phase0": { "status": "completed", "completedAt": "...", "envCount": 3 },
    "phase1": { "status": "completed", "completedAt": "...", "lakehouseCount": 2, "shortcutCount": 87 },
    "phase2": { "status": "completed", "completedAt": "...", "notebookCount": 122, "refactoredCount": 37 },
    "phase3": { "status": "in-progress", "sjdCount": 15, "completedSjds": 8 },
    "validation": { "status": "not-started" },
    "report": { "status": "not-started" }
  },
  "reportGeneratedAt": null,
  "reportPath": null,
  "poolMappings": { ... },
  "lakehouseMappings": { ... },
  "notebookMappings": { ... },
  "sjdMappings": { ... },
  "errors": [
    { "phase": "phase2", "item": "ETL_LoadCustomers", "error": "ImportError: No module named 'custom_lib'", "resolution": "Add custom_lib to Environment" }
  ]
}

import requests, json
from datetime import datetime, timezone

def rollback_migration(state_file_path, from_phase, dry_run=True):
    """
    Roll back migrated items from Fabric workspace.
    
    Args:
        state_file_path: Path to migration-state-{uuid}.json
        from_phase: Lowest phase to roll back (0 = all, 3 = SJDs only)
        dry_run: If True, list items without deleting
    """
    with open(state_file_path) as f:
        state = json.load(f)

    fabric_token = get_token("https://api.fabric.microsoft.com")
    ws_id = state["target"]["workspaceId"]
    headers = {"Authorization": f"Bearer {fabric_token}"}

    # Build deletion plan in reverse phase order
    deletion_plan = []

    # Phase 3: SJDs (key: fabricSjdId — see Phase 3 Output schema)
    if from_phase <= 3 and "sjdMappings" in state:
        for source_name, mapping in state["sjdMappings"].items():
            if "fabricSjdId" in mapping:
                deletion_plan.append({
                    "phase": 3, "type": "SparkJobDefinition",
                    "name": source_name, "itemId": mapping["fabricSjdId"],
                    "warning": None
                })

    # Phase 2: Notebooks (key: fabricNotebookId — see Phase 2 Output schema)
    if from_phase <= 2 and "notebookMappings" in state:
        for source_name, mapping in state["notebookMappings"].items():
            if "fabricNotebookId" in mapping:
                deletion_plan.append({
                    "phase": 2, "type": "Notebook",
                    "name": source_name, "itemId": mapping["fabricNotebookId"],
                    "warning": None
                })

    # Phase 1: Lakehouses (key: lakehouseId — see Phase 1 Output schema)
    # Shortcuts are deleted automatically with their Lakehouse
    if from_phase <= 1 and "lakehouseMappings" in state:
        for source_name, mapping in state["lakehouseMappings"].items():
            if "lakehouseId" in mapping:
                phase1_info = state.get("phases", {}).get("phase1", {})
                shortcut_count = phase1_info.get("shortcutCount", 0)
                deletion_plan.append({
                    "phase": 1, "type": "Lakehouse",
                    "name": source_name, "itemId": mapping["lakehouseId"],
                    "warning": f"⚠ DESTRUCTIVE: Lakehouse + {shortcut_count} shortcuts + all OneLake data will be permanently deleted"
                })

    # Phase 0: Environments (key: fabricEnvironmentId — see Phase 0 Output schema)
    if from_phase <= 0 and "poolMappings" in state:
        for source_name, mapping in state["poolMappings"].items():
            if "fabricEnvironmentId" in mapping:
                deletion_plan.append({
                    "phase": 0, "type": "Environment",
                    "name": source_name, "itemId": mapping["fabricEnvironmentId"],
                    "warning": None
                })

    # --- Dry Run: Print plan ---
    print(f"Rollback Plan (from Phase {from_phase}):")
    print(f"  Items to delete: {len(deletion_plan)}")
    print()

    has_destructive = False
    for item in deletion_plan:
        status = "⚠ DESTRUCTIVE" if item["warning"] else "  safe"
        print(f"  [{status}] Phase {item['phase']} — {item['type']}: {item['name']}")
        if item["warning"]:
            print(f"           {item['warning']}")
            has_destructive = True

    if dry_run:
        print("\n  DRY RUN — no items deleted. Re-run with dry_run=False to execute.")
        return deletion_plan

    # --- Execute Deletion ---
    if has_destructive:
        confirm = input("\n  ⚠ This will permanently delete Lakehouse data. Type 'DELETE' to confirm: ")
        if confirm != "DELETE":
            print("  Rollback cancelled.")
            return []

    results = {"deleted": 0, "failed": 0, "errors": []}

    for item in deletion_plan:
        resp = requests.delete(
            f"https://api.fabric.microsoft.com/v1/workspaces/{ws_id}/items/{item['itemId']}",
            headers=headers
        )

        if resp.status_code in (200, 204):
            results["deleted"] += 1
            print(f"  ✓ Deleted {item['type']}: {item['name']}")
        elif resp.status_code == 404:
            results["deleted"] += 1  # Already gone
            print(f"  ✓ Already deleted: {item['name']}")
        else:
            results["failed"] += 1
            results["errors"].append({
                "item": item["name"], "type": item["type"],
                "status": resp.status_code, "error": resp.text
            })
            print(f"  ✗ Failed to delete {item['type']}: {item['name']} — {resp.status_code}")

    # --- Update State File ---
    for phase_num in range(3, from_phase - 1, -1):
        phase_key = f"phase{phase_num}"
        if phase_key in state.get("phases", {}):
            state["phases"][phase_key]["status"] = "not_started"
            state["phases"][phase_key]["rolledBackAt"] = (
                datetime.now(timezone.utc).isoformat()
            )

    with open(state_file_path, "w") as f:
        json.dump(state, f, indent=2)

    print(f"\nRollback complete: {results['deleted']} deleted, {results['failed']} failed")
    print(f"State file updated — rolled-back phases marked as not_started")

    for err in results["errors"]:
        print(f"  FAILED: {err['item']} ({err['type']}) — {err['status']}: {err['error']}")

    return results

# Dry run — see what would be deleted (safe, no changes)
rollback_migration("migration-state-abc123.json", from_phase=2, dry_run=True)

# Execute — delete Phase 2 (Notebooks) and Phase 3 (SJDs), retry from Phase 2
rollback_migration("migration-state-abc123.json", from_phase=2, dry_run=False)

# Full cleanup — delete everything and start over
rollback_migration("migration-state-abc123.json", from_phase=0, dry_run=False)