add initial powerbi assets

hooli_data_eng/assets/powerbi_assets.py

@@ -0,0 +1,36 @@
+from dagster import AssetKey, AssetSpec
+from dagster._core.definitions.asset_spec import replace_attributes
+from dagster_powerbi import (
+    load_powerbi_asset_specs,
+    DagsterPowerBITranslator,
+    build_semantic_model_refresh_asset_definition,
+)
+from dagster_powerbi.translator import PowerBIContentData
+from hooli_data_eng.powerbi_workspace import power_bi_workspace
+
+
+class MyCustomPowerBITranslator(DagsterPowerBITranslator):
+    def get_report_spec(self, data: PowerBIContentData) -> AssetSpec:
+        return super().get_report_spec(data)._replace(group_name="BI")
+
+    def get_semantic_model_spec(self, data: PowerBIContentData) -> AssetSpec:       
+        spec = super().get_semantic_model_spec(data)
+        return replace_attributes(
+            spec,
+            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")
+
+    def get_data_source_spec(self, data: PowerBIContentData) -> AssetSpec:
+        return super().get_data_source_spec(data)._replace(group_name="BI")
+
+powerbi_assets = [
+    build_semantic_model_refresh_asset_definition(resource_key="power_bi", spec=spec)
+    if spec.tags.get("dagster-powerbi/asset_type") == "semantic_model"
+    else spec
+    for spec in load_powerbi_asset_specs(
+    power_bi_workspace, dagster_powerbi_translator=MyCustomPowerBITranslator, use_workspace_scan=True)
+]

hooli_data_eng/definitions.py

@@ -1,5 +1,4 @@
 from pathlib import Path
-
 from dagster import (
     AnchorBasedFilePathMapping,
     Definitions,
@@ -12,6 +11,7 @@
 from dagster_cloud.metadata.source_code import link_code_references_to_git_if_cloud
 
 from hooli_data_eng.assets import forecasting, raw_data, marketing, dbt_assets
+from hooli_data_eng.assets.powerbi_assets import powerbi_assets
 from hooli_data_eng.assets.dbt_assets import dbt_slim_ci_job
 from hooli_data_eng.assets.marketing import check_avg_orders
 from hooli_data_eng.assets.raw_data import check_users, raw_data_schema_checks
@@ -22,7 +22,6 @@
 from hooli_data_eng.sensors.watch_s3 import watch_s3_sensor
 from hooli_data_eng.assets.marketing import avg_orders_freshness_check, min_order_freshness_check, min_order_freshness_check_sensor, check_avg_orders, avg_orders_freshness_check_schedule
 from hooli_data_eng.assets.dbt_assets import weekly_freshness_check, weekly_freshness_check_sensor
-
 # ---------------------------------------------------
 # Assets
 
@@ -59,6 +58,7 @@
 
 marketing_assets = load_assets_from_package_module(marketing, group_name="MARKETING")
 
+
 # ---------------------------------------------------
 # Definitions
 
@@ -70,7 +70,7 @@
         {"max_concurrent": 3}
     ),  
     assets=link_code_references_to_git_if_cloud(
-        with_source_code_references([*dbt_assets, *raw_data_assets, *forecasting_assets, *marketing_assets]),
+        with_source_code_references([*dbt_assets, *raw_data_assets, *forecasting_assets, *marketing_assets, *powerbi_assets]),
         file_path_mapping=AnchorBasedFilePathMapping(
             local_file_anchor=Path(__file__),
             file_anchor_path_in_repository="hooli_data_eng/definitions.py"
@@ -89,3 +89,6 @@
     ],
     jobs=[analytics_job, predict_job, dbt_slim_ci_job],
 )
+
+
+#defs = Definitions.merge(static_defs, powerbi_assets.build_defs(dagster_powerbi_translator=MyCustomPowerBITranslator, enable_refresh_semantic_models=True))

hooli_data_eng/powerbi_workspace.py

@@ -0,0 +1,12 @@
+from dagster import EnvVar
+from dagster_powerbi import PowerBIWorkspace, PowerBIServicePrincipal
+
+# Connect using a service principal
+power_bi_workspace = PowerBIWorkspace(
+    credentials=PowerBIServicePrincipal(
+        client_id=EnvVar("AZURE_POWERBI_CLIENT_ID"),
+        client_secret=EnvVar("AZURE_POWERBI_CLIENT_SECRET"),
+        tenant_id=EnvVar("AZURE_POWERBI_TENANT_ID"),
+    ),
+    workspace_id=EnvVar("AZURE_POWERBI_WORKSPACE_ID"),
+)

hooli_data_eng/resources/__init__.py

@@ -22,6 +22,7 @@
 # from hooli_data_eng.resources.warehouse import MySnowflakeIOManager as SnowflakePandasIOManager
 from hooli_data_eng.resources.sensor_file_managers import s3FileSystem, LocalFileSystem
 from hooli_data_eng.resources.sensor_smtp import LocalEmailAlert, SESEmailAlert
+from hooli_data_eng.powerbi_workspace import power_bi_workspace
 
 from databricks.sdk import WorkspaceClient
 from dagster_databricks import PipesDatabricksClient
@@ -58,6 +59,7 @@
 
 resource_def = {
     "LOCAL": {
+        "power_bi": power_bi_workspace,
         "io_manager": DuckDBPandasIOManager(
             database=os.path.join(DBT_PROJECT_DIR, "example.duckdb")
         ),
@@ -77,6 +79,7 @@
         "pipes_k8s_client": ResourceDefinition.none_resource(),
     },
     "BRANCH": {
+        "power_bi": power_bi_workspace,
         "io_manager": SnowflakePandasIOManager(
             database="DEMO_DB2_BRANCH",
             account=EnvVar("SNOWFLAKE_ACCOUNT"),
@@ -102,6 +105,7 @@
         "pipes_k8s_client": PipesK8sClient(),
     },
     "PROD": {
+        "power_bi": power_bi_workspace,
         "io_manager": SnowflakePandasIOManager(
             database="DEMO_DB2",
             account=EnvVar("SNOWFLAKE_ACCOUNT"),

setup.py

@@ -31,6 +31,7 @@
             "requests_toolbelt",
             "html5lib",
             "scikit-learn",
+            "dagster-powerbi==0.0.11"
         ],
         extras_require={"dev": ["dagster-webserver", "pytest"],
                         "sling": ["dagster-embedded-elt"]},