diff --git a/python_modules/libraries/dagster-powerbi/dagster_powerbi/translator.py b/python_modules/libraries/dagster-powerbi/dagster_powerbi/translator.py index cbdb9f85c55d8..20822c9557c09 100644 --- a/python_modules/libraries/dagster-powerbi/dagster_powerbi/translator.py +++ b/python_modules/libraries/dagster-powerbi/dagster_powerbi/translator.py @@ -3,9 +3,13 @@ from enum import Enum from typing import Any, Dict, Literal, Optional, Sequence -from dagster import _check as check +from dagster import ( + UrlMetadataValue, + _check as check, +) from dagster._core.definitions.asset_key import AssetKey from dagster._core.definitions.asset_spec import AssetSpec +from dagster._core.definitions.metadata.metadata_set import NamespacedMetadataSet from dagster._core.definitions.metadata.metadata_value import MetadataValue from dagster._core.definitions.tags.tag_set import NamespacedTagSet from dagster._record import record @@ -99,6 +103,14 @@ def namespace(cls) -> str: return "dagster-powerbi" +class PowerBIMetadataSet(NamespacedMetadataSet): + web_url: Optional[UrlMetadataValue] = None + + @classmethod + def namespace(cls) -> str: + return "dagster-powerbi" + + class DagsterPowerBITranslator: """Translator class which converts raw response data from the PowerBI API into AssetSpecs. Subclass this class to implement custom logic for each type of PowerBI content. @@ -125,7 +137,10 @@ def get_asset_spec(self, data: PowerBIContentData) -> AssetSpec: def get_dashboard_asset_key(self, data: PowerBIContentData) -> AssetKey: return AssetKey( - ["dashboard", _clean_asset_name(_remove_file_ext(data.properties["displayName"]))] + [ + "dashboard", + _clean_asset_name(_remove_file_ext(data.properties["displayName"])), + ] ) def get_dashboard_spec(self, data: PowerBIContentData) -> AssetSpec: @@ -137,12 +152,11 @@ def get_dashboard_spec(self, data: PowerBIContentData) -> AssetSpec: for report_id in tile_report_ids ] url = data.properties.get("webUrl") - metadata = {"dagster_powerbi/web_url": MetadataValue.url(url)} if url else {} return AssetSpec( key=self.get_dashboard_asset_key(data), deps=report_keys, - metadata=metadata, + metadata={**PowerBIMetadataSet(web_url=MetadataValue.url(url) if url else None)}, tags={**PowerBITagSet(asset_type="dashboard")}, ) @@ -154,12 +168,11 @@ def get_report_spec(self, data: PowerBIContentData) -> AssetSpec: dataset_data = self.workspace_data.semantic_models_by_id.get(dataset_id) dataset_key = self.get_semantic_model_asset_key(dataset_data) if dataset_data else None url = data.properties.get("webUrl") - metadata = {"dagster_powerbi/web_url": MetadataValue.url(url)} if url else {} return AssetSpec( key=self.get_report_asset_key(data), deps=[dataset_key] if dataset_key else None, - metadata=metadata, + metadata={**PowerBIMetadataSet(web_url=MetadataValue.url(url) if url else None)}, tags={**PowerBITagSet(asset_type="report")}, ) @@ -173,12 +186,11 @@ def get_semantic_model_spec(self, data: PowerBIContentData) -> AssetSpec: for source_id in source_ids ] url = data.properties.get("webUrl") - metadata = {"dagster_powerbi/web_url": MetadataValue.url(url)} if url else {} return AssetSpec( key=self.get_semantic_model_asset_key(data), deps=source_keys, - metadata=metadata, + metadata={**PowerBIMetadataSet(web_url=MetadataValue.url(url) if url else None)}, tags={**PowerBITagSet(asset_type="semantic_model")}, ) diff --git a/python_modules/libraries/dagster-powerbi/dagster_powerbi_tests/test_translator.py b/python_modules/libraries/dagster-powerbi/dagster_powerbi_tests/test_translator.py index 40eddfe8f6a4c..de9a1e83127af 100644 --- a/python_modules/libraries/dagster-powerbi/dagster_powerbi_tests/test_translator.py +++ b/python_modules/libraries/dagster-powerbi/dagster_powerbi_tests/test_translator.py @@ -16,7 +16,7 @@ def test_translator_dashboard_spec(workspace_data: PowerBIWorkspaceData) -> None assert len(deps) == 1 assert deps[0].asset_key == AssetKey(["report", "Sales_Returns_Sample_v201912"]) assert asset_spec.metadata == { - "dagster_powerbi/web_url": MetadataValue.url( + "dagster-powerbi/web_url": MetadataValue.url( "https://app.powerbi.com/groups/a2122b8f-d7e1-42e8-be2b-a5e636ca3221/dashboards/efee0b80-4511-42e1-8ee0-2544fd44e122" ) } @@ -34,7 +34,7 @@ def test_translator_report_spec(workspace_data: PowerBIWorkspaceData) -> None: assert len(deps) == 1 assert deps[0].asset_key == AssetKey(["semantic_model", "Sales_Returns_Sample_v201912"]) assert asset_spec.metadata == { - "dagster_powerbi/web_url": MetadataValue.url( + "dagster-powerbi/web_url": MetadataValue.url( "https://app.powerbi.com/groups/a2122b8f-d7e1-42e8-be2b-a5e636ca3221/reports/8b7f815d-4e64-40dd-993c-cfa4fb12edee" ) } @@ -53,7 +53,7 @@ def test_translator_semantic_model(workspace_data: PowerBIWorkspaceData) -> None assert deps[0].asset_key == AssetKey(["data_27_09_2019_xlsx"]) assert deps[1].asset_key == AssetKey(["sales_marketing_datas_xlsx"]) assert asset_spec.metadata == { - "dagster_powerbi/web_url": MetadataValue.url( + "dagster-powerbi/web_url": MetadataValue.url( "https://app.powerbi.com/groups/a2122b8f-d7e1-42e8-be2b-a5e636ca3221/datasets/8e9c85a1-7b33-4223-9590-76bde70f9a20" ) }