Merge branch 'm-kovalsky/warehouseconstring'

Author: m-kovalsky

src/sempy_labs/__init__.py

@@ -100,6 +100,7 @@
     delete_warehouse,
     get_warehouse_columns,
     get_warehouse_tables,
+    get_warehouse_connection_string,
 )
 from ._data_pipelines import (
     list_data_pipelines,
@@ -599,4 +600,5 @@
     "get_item_definition",
     "get_workspace_network_communication_policy",
     "set_workspace_network_communication_policy",
+    "get_warehouse_connection_string",
 ]

src/sempy_labs/_warehouses.py

@@ -1,4 +1,5 @@
 from ._helper_functions import (
+    resolve_item_id,
     resolve_workspace_name_and_id,
     _base_api,
     _create_dataframe,
@@ -232,3 +233,57 @@ def get_warehouse_columns(
         )
 
     return df
+
+
+@log
+def get_warehouse_connection_string(
+    warehouse: str | UUID,
+    workspace: Optional[str | UUID] = None,
+    guest_tenant_id: Optional[UUID] = None,
+    private_link_type: Optional[str] = None,
+) -> str:
+    """
+    Returns the SQL connection string of the specified warehouse.
+
+    Parameters
+    ----------
+    warehouse : str | uuid.UUID
+        Name or ID of the Fabric warehouse.
+    workspace : str | uuid.UUID, default=None
+        The Fabric workspace name or ID.
+        Defaults to None which resolves to the workspace of the attached lakehouse
+        or if no lakehouse attached, resolves to the workspace of the notebook.
+    guest_tenant_id : uuid.UUID, default=None
+        The guest tenant ID if the end user's tenant is different from the warehouse tenant.
+    private_link_type : str, default=None
+        Indicates the type of private link this connection string uses. Must be either 'Workspace' or 'None' or left as None.
+
+    Returns
+    -------
+    str
+        Returns the SQL connection string of the specified warehouse.
+    """
+    workspace_id = resolve_workspace_id(workspace)
+    warehouse_id = resolve_item_id(
+        item=warehouse, type="Warehouse", workspace=workspace
+    )
+
+    url = f"/v1/workspaces/{workspace_id}/warehouses/{warehouse_id}/connectionString"
+
+    if private_link_type is not None and private_link_type not in ["Workspace", "None"]:
+        raise ValueError(
+            f"{icons.red_dot} private_link_type must be 'Workspace' or 'None' or left as None."
+        )
+
+    if guest_tenant_id or private_link_type:
+        params = []
+        if guest_tenant_id:
+            params.append(f"guestTenantId={guest_tenant_id}")
+        if private_link_type:
+            params.append(f"privateLinkType={private_link_type}")
+        param_str = "?" + "&".join(params)
+        url += param_str
+
+    response = _base_api(request=url, client="fabric_sp")
+
+    return response.json().get("connectionString")