Implementation:Dagster io Dagster PowerBI Workspace API

Field	Value
Implementation Name	PowerBI Workspace API
Source	python_modules/libraries/dagster-powerbi/dagster_powerbi/resource.py:L102,L393
Repository	dagster-io/dagster
Domains	Business_Intelligence, Analytics

Overview

Concrete integration for Power BI workspaces as Dagster assets provided by the dagster-powerbi library.

Description

The PowerBIWorkspace resource represents a Power BI workspace and provides utilities to interact with the Power BI API. It uses PowerBICredentials (such as PowerBIServicePrincipal) for authentication and exposes methods for fetching workspace content (reports, semantic models, dashboards, data sources). The load_powerbi_asset_specs() function converts this content into Dagster AssetSpec objects using a pluggable DagsterPowerBITranslator.

Usage

Use PowerBIWorkspace and load_powerbi_asset_specs() to bring your Power BI workspace content into Dagster's asset graph. Subclass DagsterPowerBITranslator to customize how Power BI content maps to Dagster asset keys, groups, and upstream dependencies.

Code Reference

Source Location

• PowerBIWorkspace: python_modules/libraries/dagster-powerbi/dagster_powerbi/resource.py:L102
• load_powerbi_asset_specs: python_modules/libraries/dagster-powerbi/dagster_powerbi/resource.py:L393

Key Classes

class PowerBIWorkspace(ConfigurableResource):
    """Represents a workspace in PowerBI and provides utilities
    to interact with the PowerBI API.
    """
    credentials: ResourceDependency[PowerBICredentials]
    workspace_id: str = Field(..., description="The ID of the PowerBI group to use.")
    refresh_poll_interval: int = Field(
        default=5, description="The interval in seconds to poll for refresh status."
    )
    refresh_timeout: int = Field(
        default=300,
        description="The maximum time in seconds to wait for a refresh to complete.",
    )

class PowerBIServicePrincipal(PowerBICredentials):
    client_id: str
    client_secret: str
    tenant_id: str

def load_powerbi_asset_specs(
    workspace: PowerBIWorkspace,
    dagster_powerbi_translator: Optional[
        Union[DagsterPowerBITranslator, type[DagsterPowerBITranslator]]
    ] = None,
    use_workspace_scan: bool = True,
) -> Sequence[AssetSpec]:
    """Returns a list of AssetSpecs representing the Power BI content in the workspace."""
    ...

class DagsterPowerBITranslator:
    def get_report_spec(self, data) -> AssetSpec: ...
    def get_semantic_model_spec(self, data) -> AssetSpec: ...

Import

from dagster_powerbi import PowerBIWorkspace, PowerBIServicePrincipal, DagsterPowerBITranslator, load_powerbi_asset_specs

I/O Contract

Direction	Name	Type	Description
Input	credentials	PowerBICredentials	Authentication credentials (e.g., PowerBIServicePrincipal with client_id, client_secret, tenant_id)
Input	workspace_id	str	The ID of the Power BI workspace to connect to
Input	dagster_powerbi_translator	Optional[DagsterPowerBITranslator]	Optional custom translator for mapping Power BI content to Dagster assets
Input	use_workspace_scan	bool	Whether to use admin APIs for workspace scanning (default True)
Output	asset_specs	Sequence[AssetSpec]	List of Dagster AssetSpecs representing Power BI reports, semantic models, and dashboards

Usage Examples

Basic Configuration

from dagster_powerbi import (
    PowerBIWorkspace,
    PowerBIServicePrincipal,
    load_powerbi_asset_specs,
)
import dagster as dg

workspace = PowerBIWorkspace(
    credentials=PowerBIServicePrincipal(
        client_id=dg.EnvVar("AZURE_POWERBI_CLIENT_ID"),
        client_secret=dg.EnvVar("AZURE_POWERBI_CLIENT_SECRET"),
        tenant_id=dg.EnvVar("AZURE_POWERBI_TENANT_ID"),
    ),
    workspace_id=dg.EnvVar("POWERBI_WORKSPACE_ID"),
)

pbi_specs = load_powerbi_asset_specs(workspace)

defs = dg.Definitions(
    assets=pbi_specs,
    resources={"power_bi": workspace},
)

Custom Translator

from dagster_powerbi import (
    PowerBIWorkspace,
    PowerBIServicePrincipal,
    DagsterPowerBITranslator,
    load_powerbi_asset_specs,
)
import dagster as dg

class CustomPBITranslator(DagsterPowerBITranslator):
    def get_report_spec(self, data):
        default = super().get_report_spec(data)
        return default._replace(group_name="reporting")

workspace = PowerBIWorkspace(
    credentials=PowerBIServicePrincipal(
        client_id=dg.EnvVar("AZURE_POWERBI_CLIENT_ID"),
        client_secret=dg.EnvVar("AZURE_POWERBI_CLIENT_SECRET"),
        tenant_id=dg.EnvVar("AZURE_POWERBI_TENANT_ID"),
    ),
    workspace_id=dg.EnvVar("POWERBI_WORKSPACE_ID"),
)

pbi_specs = load_powerbi_asset_specs(
    workspace, dagster_powerbi_translator=CustomPBITranslator()
)

Related Pages

• Principle:Dagster_io_Dagster_BI_Tool_Integration