Implementation:Mage ai Mage ai PowerBI Streams
| Knowledge Sources | |
|---|---|
| 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"))