clean up helper functions

src/sempy_labs/_clear_cache.py

@@ -318,8 +318,8 @@ def list_storage_account_files(
         ]
     )
 
-    onelake = _get_adls_client(storage_account)
-    fs = onelake.get_file_system_client(container)
+    client = _get_adls_client(storage_account)
+    fs = client.get_file_system_client(container)
 
     for x in list(fs.get_paths()):
         if not x.is_directory:

src/sempy_labs/_helper_functions.py

@@ -12,7 +12,6 @@
 from uuid import UUID
 import sempy_labs._icons as icons
 import urllib.parse
-from azure.core.credentials import TokenCredential, AccessToken
 import numpy as np
 from IPython.display import display, HTML
 
@@ -328,9 +327,6 @@ def get_direct_lake_sql_endpoint(
 
     from sempy_labs.tom import connect_semantic_model
 
-    (workspace_name, workspace_id) = resolve_workspace_name_and_id(workspace)
-    (dataset_name, dataset_id) = resolve_dataset_name_and_id(dataset, workspace_id)
-
     # dfP = fabric.list_partitions(dataset=dataset, workspace=workspace)
     # dfP_filt = dfP[dfP["Mode"] == "DirectLake"]
 
@@ -340,7 +336,7 @@ def get_direct_lake_sql_endpoint(
     #    )
 
     with connect_semantic_model(
-        dataset=dataset_id, readonly=True, workspace=workspace_id
+        dataset=dataset, readonly=True, workspace=workspace
     ) as tom:
         sqlEndpointId = None
         for e in tom.model.Expressions:
@@ -621,9 +617,7 @@ def _conv_b64(file):
 
 def _decode_b64(file, format: Optional[str] = "utf-8"):
 
-    result = base64.b64decode(file).decode(format)
-
-    return result
+    return base64.b64decode(file).decode(format)
 
 
 def is_default_semantic_model(
@@ -690,36 +684,31 @@ def resolve_item_type(item_id: UUID, workspace: Optional[str | UUID] = None) ->
 
 
 def resolve_dataset_from_report(
-    report: str, workspace: Optional[str | UUID] = None
+    report: str | UUID, workspace: Optional[str | UUID] = None
 ) -> Tuple[UUID, str, UUID, str]:
     """
     Obtains the basic semantic model properties from which the report's data is sourced.
 
     Parameters
     ----------
-    report : str
-        The name of the Power BI report.
+    report : str | uuid.UUID
+        The name or ID of the Power BI report.
     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.
 
     Returns
     -------
-    Tuple[UUID, str, UUID, str]
+    Tuple[uuid.UUID, str, uuid.UUID, str]
         The semantic model UUID, semantic model name, semantic model workspace UUID, semantic model workspace name
     """
 
-    (workspace_name, workspace_id) = resolve_workspace_name_and_id(workspace)
+    from sempy_labs.report._generate_report import _get_report
 
-    dfR = fabric.list_reports(workspace=workspace_id)
-    dfR_filt = dfR[dfR["Name"] == report]
-    if len(dfR_filt) == 0:
-        raise ValueError(
-            f"{icons.red_dot} The '{report}' report does not exist within the '{workspace_name}' workspace."
-        )
-    dataset_id = dfR_filt["Dataset Id"].iloc[0]
-    dataset_workspace_id = dfR_filt["Dataset Workspace Id"].iloc[0]
+    dfR = _get_report(report=report, workspace=workspace)
+    dataset_id = dfR["Dataset Id"].iloc[0]
+    dataset_workspace_id = dfR["Dataset Workspace Id"].iloc[0]
     dataset_workspace = fabric.resolve_workspace_name(dataset_workspace_id)
     dataset_name = resolve_dataset_name(
         dataset_id=dataset_id, workspace=dataset_workspace
@@ -992,34 +981,15 @@ def resolve_deployment_pipeline_id(deployment_pipeline: str) -> UUID:
     return deployment_pipeline_id
 
 
-class FabricTokenCredential(TokenCredential):
-
-    def get_token(
-        self,
-        scopes: str,
-        claims: Optional[str] = None,
-        tenant_id: Optional[str] = None,
-        enable_cae: bool = False,
-        **kwargs: any,
-    ) -> AccessToken:
-
-        import notebookutils
-
-        token = notebookutils.credentials.getToken(scopes)
-        access_token = AccessToken(token, 0)
-
-        return access_token
-
-
 def _get_adls_client(account_name):
 
+    import notebookutils
     from azure.storage.filedatalake import DataLakeServiceClient
 
     account_url = f"https://{account_name}.dfs.core.windows.net"
+    token = notebookutils.credentials.getToken(audience="storage")
 
-    service_client = DataLakeServiceClient(
-        account_url, credential=FabricTokenCredential()
-    )
+    service_client = DataLakeServiceClient(account_url, credential=token)
 
     return service_client
 

src/sempy_labs/report/_generate_report.py

@@ -8,11 +8,13 @@
     _conv_b64,
     resolve_report_id,
     resolve_dataset_name_and_id,
+    resolve_item_name_and_id,
     lro,
 )
 import sempy_labs._icons as icons
 from sempy._utils._log import log
 from uuid import UUID
+from sempy.fabric.exceptions import FabricHTTPException
 
 
 def create_report_from_reportjson(
@@ -371,3 +373,41 @@ def _create_report(
             report_workspace=report_workspace,
             dataset_workspace=dataset_workspace,
         )
+
+
+def _get_report(
+    report: str | UUID, workspace: Optional[str | UUID] = None
+) -> pd.DataFrame:
+
+    (workspace_name, workspace_id) = resolve_workspace_name_and_id(workspace)
+    (report_name, report_id) = resolve_item_name_and_id(
+        item=report, type="Report", workspace=workspace
+    )
+
+    client = fabric.FabricRestClient()
+    response = client.get(f"/v1.0/myorg/groups/{workspace_id}/reports/{report_id}")
+
+    if response.status_code != 200:
+        raise FabricHTTPException(response)
+
+    result = response.json()
+
+    new_data = {
+        "Id": result.get("id"),
+        "Report Type": result.get("reportType"),
+        "Name": result.get("name"),
+        "Web Url": result.get("webUrl"),
+        "Embed Url": result.get("embedUrl"),
+        "Is From Pbix": result.get("isFromPbix"),
+        "Is Owned By Me": result.get("isOwnedByMe"),
+        "Dataset Id": result.get("datasetId"),
+        "Dataset Workspace Id": result.get("datasetWorkspaceId"),
+        "Users": result.get("users") if result.get("users") is not None else [],
+        "Subscriptions": (
+            result.get("subscriptions")
+            if result.get("subscriptions") is not None
+            else []
+        ),
+    }
+
+    return pd.DataFrame([new_data])