add link to descriptions

dagster-io · Oct 31, 2024 · a2908f2 · a2908f2
1 parent 25a3319
commit a2908f2
Showing 1 changed file with 18 additions and 4 deletions.
diff --git a/hooli_data_eng/assets/powerbi_assets.py b/hooli_data_eng/assets/powerbi_assets.py
@@ -1,4 +1,4 @@
-from dagster import AssetKey, AssetSpec
+from dagster import AssetKey, AssetSpec, EnvVar
 from dagster._core.definitions.asset_spec import replace_attributes
 from dagster_powerbi import (
     load_powerbi_asset_specs,
@@ -11,21 +11,35 @@
 
 class MyCustomPowerBITranslator(DagsterPowerBITranslator):
     def get_report_spec(self, data: PowerBIContentData) -> AssetSpec:
-        return super().get_report_spec(data)._replace(group_name="BI")
+        spec = super().get_report_spec(data)
+        return replace_attributes(
+            spec,
+            description=f"Report link: https://app.powerbi.com/groups/{EnvVar("AZURE_POWERBI_WORKSPACE_ID").get_value()}/reports/{data.properties["id"]}",
+            group_name="BI",
+            tags={"core_kpis":"","dagster-powerbi/asset_type": "report"})
 
     def get_semantic_model_spec(self, data: PowerBIContentData) -> AssetSpec:       
         spec = super().get_semantic_model_spec(data)
         return replace_attributes(
             spec,
+            description=f"Semantic model link: https://app.powerbi.com/groups/{EnvVar("AZURE_POWERBI_WORKSPACE_ID").get_value()}/datasets/{data.properties["id"]}/details",
             group_name="BI",
             deps=[AssetKey(path=[dep.asset_key.path[1].upper(), dep.asset_key.path[2]]) for dep in spec.deps],
             tags={"core_kpis":"","dagster-powerbi/asset_type": "semantic_model"})
 
     def get_dashboard_spec(self, data: PowerBIContentData) -> AssetSpec:
-        return super().get_dashboard_spec(data)._replace(group_name="BI")
+        spec = super().get_dashboard_spec(data)
+        return replace_attributes(
+            spec,
+            group_name="BI"
+        )
 
     def get_data_source_spec(self, data: PowerBIContentData) -> AssetSpec:
-        return super().get_data_source_spec(data)._replace(group_name="BI")
+        spec = super().get_data_source_spec(data)
+        return replace_attributes(
+            spec,
+            group_name="BI"
+        )
 
 powerbi_assets = [
     build_semantic_model_refresh_asset_definition(resource_key="power_bi", spec=spec)