Implementation:Mage ai Mage ai PowerBI Streams

Knowledge Sources	Mage Integrations Power BI REST API
Domains	Data_Integration, PowerBI, API
Last Updated	2026-02-09 00:00 GMT

Overview

Defines all Power BI REST API stream classes and their sync logic for the Mage Power BI source connector.

Description

This module contains the stream class hierarchy for the Power BI source connector. The BaseStream base class provides shared attributes (tap_stream_id, replication_method, replication_key, key_properties, path, params, data_key) and abstract methods for record retrieval. Power BI API responses wrap data under the value key, which is set as the default_data_key. Two intermediate classes set replication behavior: IncrementalStream and FullTableStream. Twelve concrete stream classes represent Power BI resources in paired list/detail patterns: DashboardList/Dashboards, TileList/Tiles, ReportList/Reports, DatasetList/Datasets, AppList/Apps, and GroupList/Groups. List streams serve as parents that provide resource IDs for their detail stream counterparts.

Usage

Imported by the Power BI source connector to discover available resources, build stream objects, and execute sync operations against the Power BI REST API.

Code Reference

Source Location

Repository: mage-ai
File: mage_integrations/mage_integrations/sources/powerbi/streams.py
Lines: 1-365

Signature

class BaseStream:
    tap_stream_id = None
    replication_method = None
    replication_key = None
    key_properties = []
    path = None
    default_data_key = "value"
    data_key = None
    parent = None
    def __init__(self, client: PowerbiClient, logger=LOGGER): ...
    def get_records(self, bookmark_datetime=None, is_parent=False) -> list: ...
    def get_parent_data(self, bookmark_datetime=None) -> list: ...

class IncrementalStream(BaseStream):
    replication_method = "INCREMENTAL"

class FullTableStream(BaseStream):
    replication_method = "FULL_TABLE"

Import

from mage_integrations.sources.powerbi.streams import STREAMS, BaseStream

I/O Contract

Inputs

Name	Type	Required	Description
client	PowerbiClient	Yes	Authenticated Power BI API client
bookmark_datetime	datetime	No	Bookmark datetime for incremental sync

Outputs

Name	Type	Description
records	list[dict]	List of record dictionaries extracted from Power BI API responses

Stream Classes

Stream Name	Class	Replication	Key Properties	Parent
dashboard_list	DashboardList	FULL_TABLE	[id]	None
dashboards	Dashboards	FULL_TABLE	[id]	DashboardList
tile_list	TileList	FULL_TABLE	[id]	Dashboards
tiles	Tiles	FULL_TABLE	[id]	TileList
report_list	ReportList	FULL_TABLE	[id]	None
reports	Reports	FULL_TABLE	[id]	ReportList
dataset_list	DatasetList	FULL_TABLE	[id]	None
datasets	Datasets	FULL_TABLE	[id]	DatasetList
app_list	AppList	FULL_TABLE	[id]	None
apps	Apps	FULL_TABLE	[id]	AppList
group_list	GroupList	FULL_TABLE	[id]	None
groups	Groups	FULL_TABLE	[id]	GroupList

Key Behaviors

Value-Keyed Response Parsing

Power BI API responses return result arrays under the value key (e.g., {"value": [...]}), which differs from standard REST conventions. The default_data_key = "value" attribute ensures the sync logic correctly extracts records from this nested structure.

Paired List/Detail Pattern

Streams are organized in list/detail pairs. The list stream (e.g., DashboardList) fetches summary records from the collection endpoint and serves as a parent. The detail stream (e.g., Dashboards) uses IDs from the parent to fetch individual resource details, enabling hierarchical data extraction.

Parent Data Resolution

Each child stream's get_parent_data method instantiates its parent class with the same client and calls get_records(is_parent=True) to retrieve the parent IDs needed for constructing detail API endpoint URLs.

Usage Examples

from mage_integrations.sources.powerbi.streams import STREAMS
from mage_integrations.sources.powerbi.client import PowerbiClient

client = PowerbiClient(config={"access_token": "eyJ0eXAi..."})

# Instantiate a stream
dashboard_list_cls = STREAMS["dashboard_list"]
dashboard_stream = dashboard_list_cls(client)

# Retrieve records
dashboards = dashboard_stream.get_records()
for dashboard in dashboards:
    print(dashboard["id"], dashboard.get("displayName"))

Related Pages

Implements Principle

Principle:Mage_ai_Mage_ai_API_Stream_Discovery

Requires Environment

Environment:Mage_ai_Mage_ai_Python_3_9_Runtime

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment