I use PowerBI REST API through Python. Here's some sample code, using the `dlt` data ingestion library:
from collections.abc import Generator
from time import sleep
from typing import Any
from azure.identity import ClientSecretCredential
import dlt
from dlt.common import logger as dlt_logger
from dlt.sources.helpers.rest_client import RESTClient
from dlt.sources.helpers.rest_client.auth import OAuth2AuthBase
from dlt.sources.helpers.rest_client.paginators import SinglePagePaginator
class PowerBIOauthClientCredentials(OAuth2AuthBase):
def __init__(self, tenant_id: str, client_id: str, client_secret: str):
"""PowerBI OAuth2 client credentials authentication.
Args:
tenant_id (str): The Azure tenant ID.
client_id (str): The client ID of the service principal app.
client_secret (str): The client secret of the service principal app.
"""
super().__init__()
self.access_token = self._get_token(tenant_id, client_id, client_secret)
@staticmethod
def _get_token(tenant_id: str, client_id: str, client_secret: str) -> str:
scope = "https://analysis.windows.net/powerbi/api/.default"
client_secret_credential_class = ClientSecretCredential(
tenant_id=tenant_id, client_id=client_id, client_secret=client_secret
)
return client_secret_credential_class.get_token(scope).token
client = RESTClient(
base_url="https://api.powerbi.com/v1.0/myorg",
auth=PowerBIOauthClientCredentials(
tenant_id=tenant_id,
client_id=client_id,
client_secret=client_secret,
),
paginator=SinglePagePaginator(),
)
def workspaces() -> Generator[list[dict[str, Any]], None, None]:
endpoint = "groups"
yield client.get(endpoint).json()["value"]
def workspaces_lineage(
workspaces: list[dict[str, Any]],
) -> Generator[list[dict[str, Any]], None, None]:
workspace_ids = [workspace["id"] for workspace in workspaces]
request_lineage_endpoint = "admin/workspaces/getInfo"
params = {
"lineage": True,
"datasourceDetails": True,
"datasetSchema": True,
"datasetExpressions": True,
"getArtifactUsers": True,
}
body = {"workspaces": workspace_ids}
# Request a workspace lineage scan and await scan completion.
response = client.post(request_lineage_endpoint, params=params, json=body)
scan_id = response.json()["id"]
scan_status = None
dlt_logger.info("Waiting for scan to complete...")
while scan_status != "Succeeded":
scan_status_endpoint = f"admin/workspaces/scanStatus/{scan_id}"
scan_status = client.get(scan_status_endpoint).json()["status"]
sleep(0.1)
# Get the scan result.
scan_result_endpoint = f"admin/workspaces/scanResult/{scan_id}"
response = client.get(scan_result_endpoint)
lineage = response.json()
yield lineage
The process is this:
1. Get a list of workspaces via a GET request to https://api.powerbi.com/v1.0/myorg/groups
2. Perform a POST request to https://api.powerbi.com/v1.0/myorg/admin/workspaces/getInfo to initialize a lineage scan
3. Monitor scan status by periodically sending a GET request to https://api.powerbi.com/v1.0/myorg/admin/workspaces/scanStatus/{scan_id}
4. Once the scan is complete, send a GET request to https://api.powerbi.com/v1.0/myorg/admin/workspaces/scanResult to retrieve the lineage JSON.
From there, if you have the PowerBI settings I mentioned previously and corrent accesses configured, you will have the raw queries in the resonse. They look something like this:
"expression": "let\n Source = PostgreSQL.Database(\"some-addr.us-east-2.aws.neon.tech\", \"mydb\"),\n my_schema_my_table = Source{[Schema=\"my_schema\",Item=\"my_db\"]}[Data]\nin\n my_schema_my_table"
