1.8.0rc1

dbt/adapters/fabricspark/shortcut.py

@@ -0,0 +1,57 @@
+from dataclasses import dataclass
+from typing import Optional
+from enum import Enum
+
+class TargetName(Enum):
+    onelake = "onelake"
+
+@dataclass
+class Shortcut:
+    """
+    A shortcut that can be created for different target systems (onelake).
+
+    Attributes:
+        path (str): The path where the shortcut will be created.
+        name (str): The name of the shortcut.
+        target (TargetName): The target system where the shortcut will be created -- only 'onelake' is supported for now.
+        source_path (Optional[str]): The source path for the shortcut ('onelake' target).
+        source_workspace_id (Optional[str]): The source workspace ID for the shortcut ('onelake' target).
+        source_item_id (Optional[str]): The source item ID for the shortcut ('onelake' target).
+        location (Optional[str]): The location for the shortcut ('amazonS3' and 'adlsGen2' targets).
+        subpath (Optional[str]): The subpath for the shortcut ('amazonS3' and 'adlsGen2' targets).
+        connection_id (Optional[str]): The connection ID for the shortcut ('amazonS3', 'adlsGen2', and 'dataverse' targets).
+        delta_lake_folder (Optional[str]): The delta lake folder for the shortcut ('dataverse' target).
+        environment_domain (Optional[str]): The environment domain for the shortcut ('dataverse' target).
+        table_name (Optional[str]): The table name for the shortcut ('dataverse' target).
+    """
+    # the path where the shortcut will be created
+    path: str = None
+    shortcut_name: str = None
+    target: TargetName = None
+    endpoint: str = None
+    # onelake specific
+    source_path: Optional[str] = None
+    source_workspace_id: Optional[str] = None
+    source_item_id: Optional[str] = None
+
+    def __post_init__(self):
+        if self.path is None:
+            raise ValueError("destination_path is required")
+        if self.shortcut_name is None:
+            raise ValueError("name is required")
+        if self.target not in TargetName:
+            raise ValueError("target must be one of 'onelake', 'amazonS3', 'adlsGen2', or 'dataverse'")
+
+        if self.target == TargetName.onelake:
+            if self.source_path is None:
+                raise ValueError(f"source_path is required for {self.target}")
+            if self.source_workspace_id is None:
+                raise ValueError(f"source_workspace_id is required for {self.target}")
+            if self.source_item_id is None:
+                raise ValueError(f"source_item_id is required for {self.target}")
+
+    def __str__(self):
+        """
+        Returns a string representation of the Shortcut object.
+        """
+        return f"Shortcut: {self.shortcut_name} from {self.source_path} to {self.path}"

dbt/adapters/fabricspark/shortcut_client.py

@@ -0,0 +1,130 @@
+import requests
+import json
+from dbt.adapters.events.logging import AdapterLogger
+from dbt.adapters.fabricspark.shortcut import Shortcut, TargetName
+
+
+logger = AdapterLogger("Microsoft Fabric-Spark")
+
+class ShortcutClient:
+    def __init__(self, token: str, workspace_id: str, item_id: str, endpoint: str, shortcuts: list):
+        """
+        Initializes a ShortcutClient object.
+
+        Args:
+            token (str): The API token to use for creating shortcuts.
+            workspace_id (str): The workspace ID to use for creating shortcuts.
+            item_id (str): The item ID to use for creating shortcuts.
+            endpoint (str): Base URL of fabric api
+        """
+        self.token = token
+        self.workspace_id = workspace_id
+        self.item_id = item_id
+        self.endpoint = endpoint
+        self.shortcuts = shortcuts
+
+    def connect_url(self, shortcut:Shortcut):
+        """
+        Returns the connect URL for the shortcut.
+        """
+        return f"{self.endpoint}/workspaces/{shortcut.source_workspace_id}/items/{shortcut.source_item_id}/shortcuts/{shortcut.source_path}/{shortcut.shortcut_name}"
+
+    def get_target_body(shortcut: Shortcut):
+        """
+        Returns the target body for the shortcut based on the target attribute.
+        """
+        if shortcut.target == TargetName.onelake:
+            return {
+                shortcut.target.value: {
+                    "workspaceId": shortcut.source_workspace_id,
+                    "itemId": shortcut.source_item_id,
+                    "path": shortcut.source_path
+                }
+            }
+
+    def create_shortcuts(self, max_retries: int = 3):
+        """
+        Creates shortcuts from a JSON file.
+
+        Args:
+            retry (bool): Whether to retry creating shortcuts if there is an error (default: True).
+        """
+        for shortcut in self.shortcuts:
+            logger.debug(f"Creating shortcut: {shortcut}")
+            while max_retries > 0:
+                try:
+                    self.create_shortcut(shortcut)
+                    break
+                except Exception as e:
+                    logger.debug(f"Failed to create shortcut: {shortcut} with error: {e}. Retrying...")
+                    max_retries -= 1
+            if max_retries == 0:
+                raise f"Failed to create shortcut: {shortcut} after {max_retries} retries, failing..."
+
+    def check_exists(self, shortcut: Shortcut):
+        """
+        Checks if a shortcut exists.
+
+        Args:
+            shortcut (Shortcut): The shortcut to check.
+        """
+        headers = {
+            "Authorization": f"Bearer {self.token}",
+            "Content-Type": "application/json"
+        }
+        response = requests.get(shortcut.connect_url(), headers=headers)
+        # check if the error is ItemNotFound
+        if response.status_code == 404:
+            return False
+        response.raise_for_status() # raise an exception if there are any other errors
+        # else, check that the target body of the existing shortcut matches the target body of the shortcut they want to create
+        response_json = response.json()
+        response_target = response_json["target"]
+        target_body = shortcut.get_target_body()
+        if response_target != target_body:
+            # if the response target does not match the target body, delete the existing shortcut, then return False so we can create the new shortcut
+            logger.debug(f"Shortcut {shortcut} already exists with different source path, workspace ID, and/or item ID. Deleting exisiting shortcut and recreating based on JSON.")
+            self.delete_shortcut(response_json["path"], response_json["name"])
+            return False
+        return True
+
+    def delete_shortcut(self, shortcut_path: str, shortcut_name: str):
+        """
+        Deletes a shortcut.
+
+        Args:
+            shortcut_path (str): The path where the shortcut is located.
+            shortcut_name (str): The name of the shortcut.
+        """
+        connect_url = f"{self.endpoint}/workspaces/{self.workspace_id}/items/{self.item_id}/shortcuts/{shortcut_path}/{shortcut_name}"
+        headers = {
+            "Authorization": f"Bearer {self.token}",
+            "Content-Type": "application/json"
+        }
+        logger.debug(f"Deleting shortcut {shortcut_name} at {shortcut_path} from workspace {self.workspace_id} and item {self.item_id}")
+        response = requests.delete(connect_url, headers=headers)
+        response.raise_for_status()
+
+    def create_shortcut(self, shortcut: Shortcut):
+        """
+        Creates a shortcut.
+
+        Args:
+            shortcut (Shortcut): The shortcut to create.
+        """
+        if self.check_exists(shortcut):
+            logger.debug(f"Shortcut {shortcut} already exists, skipping...")
+            return
+        connect_url = f"{self.endpoint}/workspaces/{self.workspace_id}/items/{self.item_id}/shortcuts"
+        headers = {
+            "Authorization": f"Bearer {self.token}",
+            "Content-Type": "application/json"
+        }
+        target_body = self.get_target_body(shortcut)
+        body = {
+            "path": shortcut.path,
+            "name": shortcut.shortcut_name,
+            "target": target_body
+        }
+        response = requests.post(connect_url, headers=headers, data=json.dumps(body))
+        response.raise_for_status()

shortcuts.json

@@ -0,0 +1,18 @@
+"shortcut_paths" : [
+	{
+		"path": "Tables/",
+		"shortcut_table_name":"test_shortcut",
+		"target": "onelake",
+		"source_workspaceid": "asdfsdf",
+		"source_itemid": "asdfsadf",
+		"source_data_path":"asdfsdfs"
+	},
+	{
+		"path": "Tables/",
+		"shortcut_table_name":"",
+		"target": "onelake",
+		"source_workspaceid": "",
+		"source_itemid": "",
+		"source_data_path":""
+	}
+]