Forum Discussion
MathieuSGA
1 year agoAdvocate I
'InsufficientScopes' error while trying to create shortcut through Fabric REST API
Hi, Few weeks ago, I was using the following process without any trouble but can't seem to figure out what went wrong. Any changes in permission ? credentials ? I'm not particular at ease with ...
- 1 year ago
Something definitely has changed in Fabric since yesterday. I faced a similar issue yesterday while retrieving connections information through FabricRestClient. The code was working fine until the previous day.
The way I got around it to explicitly initialize token using notebookutils.getToken and pass that in request headers. Your code does this for list_shortcuts whereas for create_sbortcuts it is not passing the explicit header. Initialize and add the request header to your create shortcuts method and it may work.
VenDaFabricator
1 year agoResolver I
Here is full code: with modifications as suggested by gaya3krishnan86
def createshortcuts(source_workspaceid, source_lakehouseid, source_folder_path,
destination_workspaceid, destination_lakehouseid, destination_folder_path,
searchPattern):
SOURCE_URI = f"abfss://{source_workspaceid}@onelake.dfs.fabric.microsoft.com/{source_lakehouseid}/{source_folder_path}"
DEST_URI = f"abfss://{destination_workspaceid}@onelake.dfs.fabric.microsoft.com/{destination_lakehouseid}/{destination_folder_path}"
PATTERN_MATCH = searchPattern
client = FabricRestClient()
def extract_onelake_https_uri_components(uri):
pattern = re.compile(r"abfss://([^@]+)@[^/]+/([^/]+)/(.*)")
match = pattern.search(uri)
if match:
workspace_id, item_id, path = match.groups()
return workspace_id, item_id, path
else:
return None, None, None
def is_valid_onelake_uri(uri: str) -> bool:
workspace_id, item_id, path = extract_onelake_https_uri_components(uri)
return all([workspace_id, item_id, path])
def get_last_path_segment(uri: str):
return uri.split("/")[-1] if uri else None
def is_delta_table(uri: str):
delta_log_path = os.path.join(uri, "_delta_log")
return mssparkutils.fs.exists(delta_log_path)
def is_folder_matching_pattern(path: str, folder_name: str, patterns: []):
if folder_name in patterns:
return True
for pattern in patterns:
if fnmatch.fnmatch(folder_name, pattern):
return is_delta_table(path)
return False
def get_matching_delta_tables_uris(uri: str, patterns: []) -> []:
matched_uris = set()
try:
files = mssparkutils.fs.ls(uri)
folders = [item for item in files if item.isDir]
matched_uris.update(
folder.path
for folder in folders
if is_folder_matching_pattern(folder.path, folder.name, patterns)
)
except Exception as e:
print(f"❌ Error listing folders at {uri}: {str(e)}")
return matched_uris
def get_onelake_shorcut(workspace_id: str, item_id: str, path: str, name: str):
shortcut_uri = f"v1/workspaces/{workspace_id}/items/{item_id}/shortcuts/{path}/{name}"
try:
return client.get(shortcut_uri).json()
except Exception as e:
print(f"⚠️ Failed to retrieve shortcut metadata: {e}")
return None
def create_onelake_shorcut(source_uri: str, dest_uri: str):
src_workspace_id, src_item_id, src_path = extract_onelake_https_uri_components(source_uri)
dest_workspace_id, dest_item_id, dest_path = extract_onelake_https_uri_components(dest_uri)
name = get_last_path_segment(source_uri)
dest_uri_joined = os.path.join(dest_uri, name)
if mssparkutils.fs.exists(dest_uri_joined):
print(f"⚠️ Destination already exists: {dest_uri_joined}")
return None
request_body = {
"name": name,
"path": dest_path,
"target": {
"oneLake": {
"itemId": src_item_id,
"path": src_path,
"workspaceId": src_workspace_id,
}
},
}
shortcut_uri = f"https://api.fabric.microsoft.com/v1/workspaces/{dest_workspace_id}/items/{dest_item_id}/shortcuts"
# ✅ Explicit token for new Fabric security model
token = mssparkutils.credentials.getToken("https://api.fabric.microsoft.com")
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
print(f"📎 Creating shortcut: {shortcut_uri}/{name}...")
try:
response = requests.post(shortcut_uri, headers=headers, json=request_body)
if response.status_code not in [200, 201]:
print(f"❌ Failed to create shortcut. Status: {response.status_code}")
print(f"Message: {response.text}")
return None
except Exception as e:
print(f"❌ Exception while creating shortcut: {str(e)}")
return None
return response.json()
# Validate input
if PATTERN_MATCH is None or len(PATTERN_MATCH) == 0:
raise TypeError("Argument 'searchPattern' should be a valid non-empty list")
if not is_valid_onelake_uri(SOURCE_URI) or not is_valid_onelake_uri(DEST_URI):
print("❌ Invalid URIs. Ensure URIs are in abfss://<workspace-id>@onelake.dfs.fabric.microsoft.com/<lakehouse-id>/<path> format.")
return []
source_uri_addr = SOURCE_URI.rstrip("/")
dest_uri_addr = DEST_URI.rstrip("/")
dest_workspace_id, dest_item_id, dest_path = extract_onelake_https_uri_components(dest_uri_addr)
result = []
if not dest_path.startswith("Tables") or is_delta_table(source_uri_addr):
shortcut = create_onelake_shorcut(source_uri_addr, dest_uri_addr)
if shortcut:
result.append(shortcut)
else:
for delta_table_uri in get_matching_delta_tables_uris(source_uri_addr, PATTERN_MATCH):
shortcut = create_onelake_shorcut(delta_table_uri, dest_uri_addr)
if shortcut:
result.append(shortcut)source_workspaceid = d76d5xxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
source_lakehouseid = 8a2f8xxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
destination_workspaceid = d76d5xxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
destination_lakehouseid = ee5d0xxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
destination_folder_path = 'Tables/Bronze'
searchPattern = [ 'customer', 'location', 'products']
createshortcuts(source_workspaceid, source_lakehouseid, source_folder_path, destination_workspaceid, destination_lakehouseid, destination_folder_path, searchPattern)