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 this topic, so I'm really looking forward to hearing from you experts !
# Issue description
In one Microsoft Fabric ws_feat workspace, I had:
- one lib_Shortcut Notebook where I defined a '
FabricShortcutManager' class to help me to create shortcuts from other notebooks- one NB_create_shortcut notebook where I was "calling" the lib_Shortcut. Now, when using the 'FabricShortcutManager.create_shortcuts_from_pattern', which actually implement...
```
shortcut_uri = f"v1/workspaces/{dest_workspace_id}/items/{dest_item_id}/shortcuts"
self.client.post(shortcut_uri, json=request_body)
```
...that lead me into an 'InsufficientScopes' Error
# Configuration
## FabricShortcutManager' class
class FabricShortcutManager:
def __init__(self, destination_workspace_id, destination_lakehouse_id):
# https://learn.microsoft.com/en-us/python/api/semantic-link-sempy/sempy.fabric?view=semantic-link-python
# FabricRestClient uses the default credentials of the executing user
self.client = fabric.FabricRestClient()
self.request_headers = {
"Authorization": "Bearer " + mssparkutils.credentials.getToken("pbi"),
"Content-Type": "application/json"
}
self.destination_workspace_id = destination_workspace_id
self.destination_lakehouse_id = destination_lakehouse_id
# Extract workspace_id, item_id and path from a onelake URI
def extract_onelake_https_uri_components(self, uri):
# Define a regular expression to match any string between slashes and capture the final path element(s) without the leading slash
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(self, uri: str) -> bool:
workspace_id, item_id, path = self.extract_onelake_https_uri_components(uri)
if "abfss://" not in uri or workspace_id is None or item_id is None or path is None:
return False
return True
def get_last_path_segment(self, uri: str):
path = uri.split("/") # Split the entire URI by '/'
return path[-1] if path else None
def is_delta_table(self, uri: str):
delta_log_path = os.path.join(uri, "_delta_log")
return mssparkutils.fs.exists(delta_log_path)
def is_folder_matching_pattern(self, path: str, folder_name: str, patterns: []):
if folder_name in patterns:
return True
else:
for pattern in patterns:
if fnmatch.fnmatch(folder_name, pattern):
return self.is_delta_table(path)
return False
def get_matching_delta_tables_uris(self, uri: str, patterns: []) -> []:
# Use a set to avoid duplicates
matched_uris = set()
files = mssparkutils.fs.ls(uri)
folders = [item for item in files if item.isDir]
# Filter folders to only those that matches the pattern and is a delta table
matched_uris.update(
folder.path
for folder in folders
if self.is_folder_matching_pattern(folder.path, folder.name, patterns)
)
return matched_uris
def get_onelake_shorcut(self, workspace_id: str, item_id: str, path: str, name: str):
shortcut_uri = (
f"v1/workspaces/{workspace_id}/items/{item_id}/shortcuts/{path}/{name}"
)
response = self.client.get(shortcut_uri).json()
return response
def delete_onelake_shorcut(self, workspace_id: str, item_id: str, path: str, name: str):
shortcut_uri = (
f"v1/workspaces/{workspace_id}/items/{item_id}/shortcuts/{path}/{name}"
)
response = self.client.delete(shortcut_uri)
if response.status_code == 200:
# Wait for the delete operation to fully propogate
shortcut_uri = f"abfss://{workspace_id}@onelake.dfs.fabric.microsoft.com/{item_id}/{path}/{name}"
while mssparkutils.fs.exists(shortcut_uri):
time.sleep(5)
return response
def create_onelake_shorcut(self, source_uri: str, dest_uri: str):
src_workspace_id, src_item_id, src_path = self.extract_onelake_https_uri_components(
source_uri
)
dest_workspace_id, dest_item_id, dest_path = self.extract_onelake_https_uri_components(
dest_uri
)
name = self.get_last_path_segment(source_uri)
dest_uri_joined = os.path.join(dest_uri, name)
# If the destination path already exists, return without creating shortcut
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"v1/workspaces/{dest_workspace_id}/items/{dest_item_id}/shortcuts"
print(f"Creating shortcut: {shortcut_uri}/{name}..")
print(f"{request_body=}")
try:
self.client.post(shortcut_uri, json=request_body)
except FabricHTTPException as e:
print(e)
return None
return self.get_onelake_shorcut(dest_workspace_id, dest_item_id, dest_path, name)
def list_shortcuts(self, workspace_id: str = None, item_id: str = None):
destination_workspace_id = workspace_id or self.destination_workspace_id
destination_lakehouse_id = item_id or self.destination_lakehouse_id
shortcut_uri = f"v1/workspaces/{destination_workspace_id}/items/{destination_lakehouse_id}/shortcuts"
try:
print(f"Listing shortcut: {shortcut_uri}..")
response = self.client.get(shortcut_uri, headers=self.request_headers)
return response.status_code, response # self.format_response(response)
except FabricHTTPException as e:
print(e)
return None, None
def create_shortcuts_from_pattern(self, source_uri, dest_uri, pattern_match):
"""
# URI's should be in the form: abfss://<workspace-id>@onelake.dfs.fabric.microsoft.com/<item-id>/<path>"
SOURCE_URI = "abfss://<workspace id>@onelake.dfs.fabric.microsoft.com/<item id>/<path>"
DEST_URI = "abfss://<workspace id>@onelake.dfs.fabric.microsoft.com/<item id>/<path>"
# Provide an array of search wildcards or just "*" to shortcut all tables
# A list of specific table names will also be matches specifically such as ["custtable", "salestable", "salesline"]
#
# Wildcard Syntax:
# * Matches everything
# ? Matches any single character
# [seq] Matches any character in seq
# [!seq] Matches any character not in seq
#
# Examples:
# "cust*" # Matches any table beginning with "cust"
# "*cust*" # Matches any table containing the string "cust"
# "custtabl?" Matches "custtable" but not "custtables"
# "custtable_[1-2]" Matches "custtable_1", "custtable_2" but not "custtable_3"
# "custtable_[!1-2]" Matches "custtable_3", "custtable_4" (etc.), but not "custtable_1" or "custtable_2"
PATTERN_MATCH = ["*"]
"""
if pattern_match is None or len(pattern_match) == 0:
raise TypeError("Argument 'pattern_match' should be a valid list of patterns or ["*"] to match everything")
# Collect created shortcuts
result = []
# If either URI's are invalid, just return
if not self.is_valid_onelake_uri(source_uri) or not self.is_valid_onelake_uri(dest_uri):
print(
"invalid URI's provided. URI's should be in the form: abfss://<workspace-id>@onelake.dfs.fabric.microsoft.com/<item-id>/<path>"
)
else:
# Remove any trailing '/' from uri's
source_uri_addr = source_uri.rstrip("/")
dest_uri_addr = dest_uri.rstrip("/")
dest_workspace_id, dest_item_id, dest_path = self.extract_onelake_https_uri_components(
dest_uri_addr
)
# If we are not shortcutting to a managed table folder or
# the source uri is a delta table, just shortcut it 1-1.
if not dest_path.startswith("Tables") or self.is_delta_table(source_uri_addr):
shortcut = self.create_onelake_shorcut(source_uri_addr, dest_uri_addr)
if shortcut is not None:
result.append(shortcut)
else:
# If source is not a delta table, and destination is managed table folder:
# Iterate over source folders and create table shortcuts @ destination
for delta_table_uri in self.get_matching_delta_tables_uris(
source_uri_addr, pattern_match
):
shortcut = self.create_onelake_shorcut(delta_table_uri, dest_uri_addr)
if shortcut is not None:
result.append(shortcut)
return result
def delete_shortcuts_from_pattern(self, source_uri, dest_uri, pattern_match):
"""
# URI's should be in the form: abfss://<workspace-id>@onelake.dfs.fabric.microsoft.com/<item-id>/<path>"
DEST_URI = "abfss://<workspace id>@onelake.dfs.fabric.microsoft.com/<item id>/<path>"
# Provide an array of search wildcards or just "*" to shortcut all tables
# A list of specific table names will also be matches specifically such as ["custtable", "salestable", "salesline"]
#
# Wildcard Syntax:
# * Matches everything
# ? Matches any single character
# [seq] Matches any character in seq
# [!seq] Matches any character not in seq
#
# Examples:
# "cust*" # Matches any table beginning with "cust"
# "*cust*" # Matches any table containing the string "cust"
# "custtabl?" Matches "custtable" but not "custtables"
# "custtable_[1-2]" Matches "custtable_1", "custtable_2" but not "custtable_3"
# "custtable_[!1-2]" Matches "custtable_3", "custtable_4" (etc.), but not "custtable_1" or "custtable_2"
PATTERN_MATCH = ["*"]
"""
if pattern_match is None or len(pattern_match) == 0:
raise TypeError("Argument 'pattern_match' should be a valid list of patterns or ["*"] to match everything")
# Collect created shortcuts
result = []
# If either URI's are invalid, just return
if not self.is_valid_onelake_uri(dest_uri):
print(
"invalid URI's provided. URI's should be in the form: abfss://<workspace-id>@onelake.dfs.fabric.microsoft.com/<item-id>/<path>"
)
else:
# Remove any trailing '/' from uri's
dest_uri_addr = dest_uri.rstrip("/")
dest_workspace_id, dest_item_id, dest_path = self.extract_onelake_https_uri_components(
dest_uri_addr
)
# If we are not shortcutting to a managed table folder or
# the source uri is a delta table, just shortcut it 1-1.
if not dest_path.startswith("Tables"):
raise NotImplementedError(f"Please go inside the 'FabricShortcutManager' class implementation")
else:
# If source is not a delta table, and destination is managed table folder:
# Iterate over source folders and create table shortcuts @ destination
for delta_table_uri in self.get_matching_delta_tables_uris(
dest_uri_addr, pattern_match
):
shortcut = self.delete_onelake_shorcut(dest_workspace_id, dest_item_id, dest_path)
if shortcut is not None:
result.append(shortcut)
return result
@staticmethod
def format_response(response):
# Build the return payload for a success response
if (response.status_code >= 200 and response.status_code <= 299):
response_content = {
"request_url" : response.url,
"response_content" : {} if response.text == '' else json.loads(response.text),
"status" : "success",
"status_code" : response.status_code,
"status_description" : status_codes._codes[response.status_code][0]
}
# Build the return payload for a failure response
if not (response.status_code >= 200 and response.status_code <= 299):
response_content = {
"request_body" : request_body,
"request_headers" : request_headers,
"request_url" : response.url,
"response_text" : json.loads(response.text),
"status" : "error",
"status_code" : response.status_code,
"status_description" : status_codes._codes[response.status_code][0]
}
return response_content
### class UriParser (from lib_Shortcut notebook)
class UriParser:
@classmethod
def build(cls, workspace_id, lakehouse_id, dir_path: str = None, file_name: str = None):
abfss_path = f"abfss://{workspace_id}@onelake.dfs.fabric.microsoft.com/{lakehouse_id}"
if dir_path:
abfss_path += f"/{dir_path}"
if file_name:
abfss_path += f"/{file_name}"
return abfss_path
# Extract workspace_id, item_id and path from a onelake URI
@classmethod
def extract_onelake_https_uri_components(cls, uri):
# Define a regular expression to match any string between slashes and capture the final path element(s) without the leading slash
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
@classmethod
def is_valid_onelake_uri(cls, uri: str) -> bool:
workspace_id, item_id, path = cls.extract_onelake_https_uri_components(uri)
if "abfss://" not in uri or workspace_id is None or item_id is None or path is None:
return False
return True
@classmethod
def get_last_path_segment(cls, uri: str):
path = uri.split("/") # Split the entire URI by '/'
return path[-1] if path else None
@classmethod
def is_delta_table(cls, uri: str):
delta_log_path = os.path.join(uri, "_delta_log")
return mssparkutils.fs.exists(delta_log_path)
@classmethod
def get_matching_delta_tables_uris(cls, uri: str, patterns: []) -> []:
# Use a set to avoid duplicates
matched_uris = set()
files = mssparkutils.fs.ls(uri)
folders = [item for item in files if item.isDir]
# Filter folders to only those that matches the pattern and is a delta table
matched_uris.update(
folder.path
for folder in folders
if cls.is_folder_matching_pattern(folder.path, folder.name, patterns)
)
return matched_uris
@classmethod
def is_folder_matching_pattern(cls, path: str, folder_name: str, patterns: []):
if folder_name in patterns:
return True
else:
for pattern in patterns:
if fnmatch.fnmatch(folder_name, pattern):
return cls.is_delta_table(path)
return False
## NB_create_shortcut
%run lib_Shortcut
sc_manager = FabricShortcutManager(destination_ws_id, destination_lakehouse_id)
source_uri = UriParser.build(source_ws_id, source_lakehouse_id, source_dir)
destination_uri = UriParser.build(destination_ws_id, destination_lakehouse_id, source_dir)
shortcuts_created = sc_manager.create_shortcuts_from_pattern(source_uri, destination_uri, pattern_match=['task', 'affaire'])
