Efficiently Moving Items Between Workspaces with Fabric CLI

Many people think of Fabric CLI as a developer tool for use on a local machine, and understandably so, since that's the typical view of any command-line interface (CLI). However, what if you're already immersed in a notebook, examining capabilities, scripting your solution, and you prefer not to change tools? This is precisely why I prefer Fabric CLI over any others I've used.

My Problem:

I have way too many items in a folder under a particular workspace. I need to move or copy only those items into a separate, dedicated workspace. Bonus, if I can create a folder and keep all these items in that folder.

Step 1: Authenticate Fabric CLI Inside a Notebook

First step authentication. I was initially using a service principal, but Sandeep demonstrated how he was using it differently with a token, so I updated the code.

Code for this step:

# Install Fabric CLI if not already installed
# %pip install ms-fabric-cli --quiet

import os
import notebookutils

# Step 1: Authenticate and set environment variables
# This uses the token available within the notebook context
# and makes it available for all fab CLI operations

token = notebookutils.credentials.getToken('pbi')
os.environ['FAB_TOKEN'] = token
os.environ['FAB_TOKEN_ONELAKE'] = token

print("Authentication tokens set for fab CLI")

No login prompts. No switching tools.

Step 2: Define Workspaces and Folders

We define the workspace and folder names to use throughout the script.

Code for this step:

# Define source and target workspaces and folder names
source_workspace_id = "41706d46-7c89-45c1-b56f-74aca7d99000"
target_workspace_id = "bf4c8872-f858-4359-8221-df042403c3f8"

source_folder_name = "Linkedin"
target_folder_name = "Linkedintest"

# Resolve workspace names for readable CLI paths
import sempy.fabric as fabric
source_workspace_name = fabric.resolve_workspace_name(source_workspace_id)
target_workspace_name = fabric.resolve_workspace_name(target_workspace_id)

print("Source workspace name:", source_workspace_name)
print("Target workspace name:", target_workspace_name)

Step 3: Get (or Create) Folder IDs

To work with folders, we need their internal IDs. This function fetches the ID for a given name, or creates the folder if it doesn't exist.

Code for this step:

import json
import subprocess

def get_folder_id(workspace_id, folder_name):
    cmd = f"fab api -X get workspaces/{workspace_id}/folders"
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    data = json.loads(result.stdout)
    text = data.get("text")
    folders_json = json.loads(text) if isinstance(text, str) else text
    folders = folders_json.get("value", [])
    for f in folders:
        if f.get("displayName") == folder_name:
            return f.get("id")
    return None

# Retrieve IDs or create the target folder if needed
source_folder_id = get_folder_id(source_workspace_id, source_folder_name)
if not source_folder_id:
    raise Exception(f"Source folder '{source_folder_name}' not found")

target_folder_id = get_folder_id(target_workspace_id, target_folder_name)
if not target_folder_id:
    payload = json.dumps({"displayName": target_folder_name})
    cmd_create = f"fab api -X post workspaces/{target_workspace_id}/folders -i '{payload}'"
    result_create = subprocess.run(cmd_create, shell=True, capture_output=True, text=True)
    resp = json.loads(result_create.stdout)
    target_folder_id = resp.get("text", {}).get("id")
    if not target_folder_id:
        raise Exception(f"Failed to create target folder '{target_folder_name}'")
    print(f"Created target folder '{target_folder_name}' with ID: {target_folder_id}")
else:
    print(f"Target folder '{target_folder_name}' found with ID: {target_folder_id}")

print(f"Source folder ID: {source_folder_id}")
print(f"Target folder ID: {target_folder_id}")

Step 4: List Items in Source Folder

Here we retrieve all the items inside the source folder, so we know what to work with.

Code for this step:

cmd_items = f'fab api -X get "workspaces/{source_workspace_id}/items?folderId={source_folder_id}"'
result_items = subprocess.run(cmd_items, shell=True, capture_output=True, text=True)
items_data = json.loads(result_items.stdout)
text_items = items_data.get("text")
inner_items = json.loads(text_items) if isinstance(text_items, str) else text_items
items = inner_items.get("value", [])

# Filter only items that belong directly to this folder
filtered_items = [item for item in items if item.get("folderId") == source_folder_id]
print(f"Found {len(filtered_items)} items in source folder")

Step 5: Check Before You Copy

This utility function helps prevent duplicates by checking if an item already exists in the target workspace.

Code for this step:

def check_item_exists(path):
    escaped_path = path.replace(" ", "\ ")
    cmd = f'fab exists "{escaped_path}"'
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    stdout = result.stdout.strip().lower()
    return "* true" in stdout

Step 6: Copy and Patch

This section performs the main task: copying the item and placing it in the correct folder of the destination workspace. The patch piece needs more testing. It was a bit of hit and miss.

Code for this step:

for item in filtered_items:
    item_name = item.get("displayName")
    item_type = item.get("type")

    src_path_raw = f"{source_workspace_name}.Workspace/{item_name}.{item_type}"
    tgt_path_raw = f"{target_workspace_name}.Workspace/{item_name}.{item_type}"

    if check_item_exists(tgt_path_raw):
        print(f"Skipped (exists in target): {item_name}")
        continue

    if not check_item_exists(src_path_raw):
        print(f"Skipped (missing in source): {item_name}")
        continue

    cmd_copy = f'fab cp "{src_path_raw}" "{tgt_path_raw}"'
    proc = subprocess.Popen(
        cmd_copy,
        shell=True,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )
    stdout, stderr = proc.communicate(input="Yes
")

    if proc.returncode != 0:
        print(f"Copy failed: {item_name} ({stderr.strip()})")
        continue

    get_items_cmd = f'fab api -X get "workspaces/{target_workspace_id}/items?filter=displayName eq \'{item_name}\'"'
    get_result = subprocess.run(get_items_cmd, shell=True, capture_output=True, text=True)
    get_json = json.loads(get_result.stdout)
    items_list = get_json.get("text", {}).get("value", [])

    if not items_list:
        print(f"Item not found after copy: {item_name}")
        continue

    copied_item_id = items_list[0]["id"]
    patch_body = json.dumps({"parentFolderId": target_folder_id})
    patch_cmd = f'fab api -X patch workspaces/{target_workspace_id}/items/{copied_item_id} -i \'{patch_body}\''
    patch_result = subprocess.run(patch_cmd, shell=True, capture_output=True, text=True)

    if patch_result.returncode == 0:
        print(f"Moved '{item_name}' to folder '{target_folder_name}'")
    else:
        print(f"Failed to move '{item_name}' to folder")

Every item gets checked, copied, confirmed, and moved. Neat.

Here's the full working version: GitHub Repo

Let me know if this helps — or if you've got other CLI hacks you use inside Fabric notebooks 👩🏾‍💻

#DataNovaWithPrathy #PrathyExplainsBI #DesignFirstWithPrathy