Implementing Power BI Tenant-Level Scanner API and Metadata Harvesting to Build an Enterprise-Wide Data Catalog for Cross-Workspace Asset Discovery and Governance Reporting

Introduction

Picture this: your organization has 200 Power BI workspaces spread across business units, regions, and project teams. The CFO's office wants to know which reports touch payroll data. The data governance team needs to identify every dataset that connects to the legacy ERP system before it gets decommissioned next quarter. A new GDPR audit requires you to produce a complete inventory of where customer PII appears across your analytics estate. You open the Power BI admin portal, click around for a few minutes, and quickly realize that the UI wasn't built for this kind of question. It shows you one workspace at a time.

This is exactly the problem the Power BI Scanner API was designed to solve. The Scanner API — formally part of the Admin REST API surface under the workspaces/modified and workspaces/getInfo endpoint family — lets you programmatically harvest metadata from every workspace in your tenant simultaneously. We're talking datasets, reports, dashboards, dataflows, columns, measures, tables, data sources, sensitivity labels, endorsements, and lineage relationships — all extractable as structured JSON that you can load into a catalog, a database, or a governance dashboard.

By the end of this lesson, you'll have built a complete metadata harvesting pipeline: authenticating as a service principal, triggering tenant-wide scans, polling for completion, parsing the nested JSON payload, loading it into a SQL store, and surfacing it in a Power BI governance report that answers the kinds of questions your audit team actually asks.

What you'll learn:

• How the Scanner API works under the hood — the scan lifecycle, throttling behavior, and what data it actually returns
• How to register a service principal and configure tenant settings to enable full metadata extraction including sensitive schemas
• How to write a production-grade Python harvesting script that handles pagination, polling, and error recovery
• How to design a relational schema for storing Power BI metadata and build incrementally updating catalog tables
• How to build a governance reporting layer on top of the catalog to answer cross-workspace lineage, asset inventory, and compliance questions

Prerequisites

This lesson assumes you're comfortable with:

• Power BI administration — you should understand workspaces, capacities, and tenant settings at the level covered in Power BI Governance: Workspaces, Permissions, and Audit Logging
• The Power BI REST API fundamentals. If you haven't worked with it before, review Power BI REST API: Automate Administration and Deployments first
• Python at an intermediate level — we'll use requests, msal, pandas, and sqlalchemy
• Azure Active Directory app registration and service principal concepts
• A Power BI tenant where you have Fabric Administrator or Power BI Administrator role

A Premium Per User (PPU) or Premium/Fabric capacity is required for the Scanner API's full metadata extraction (column-level detail, expressions, sensitivity labels). Basic workspace inventory works on Pro, but you'll miss most of the valuable governance data.

Understanding the Scanner API Architecture

Before writing a single line of code, you need to understand why the Scanner API works the way it does, because its design — asynchronous, batched, eventually consistent — will directly shape how you build your harvesting pipeline.

The Scan Lifecycle

The Scanner API is not a simple GET endpoint. It's an asynchronous workflow with three distinct phases:

Phase 1 — Modified Workspace Discovery. You call GET /v1.0/myorg/admin/workspaces/modified to get a list of workspace IDs that have changed since a given date. On your first run, you skip the date filter to get everything. On subsequent runs, you pass the timestamp from your last successful scan. This is your incremental update mechanism.

Phase 2 — Scan Initiation. You call POST /v1.0/myorg/admin/workspaces/getInfo with a JSON body containing a list of workspace IDs (up to 100 per request) and a set of feature flags that control what metadata gets extracted. This returns a scanId — a GUID that represents your pending scan job.

Phase 3 — Result Polling and Retrieval. You poll GET /v1.0/myorg/admin/workspaces/scanStatus/{scanId} until the status returns Succeeded. Then you call GET /v1.0/myorg/admin/workspaces/scanResult/{scanId} to retrieve the full metadata payload.

The scan typically completes in 30 seconds to 3 minutes for a batch of 100 workspaces. For a 500-workspace tenant, you're looking at 5 batches and roughly 10–15 minutes of wall-clock time if you run sequentially. We'll address parallelization later.

What the Payload Actually Contains

The scan result is a deeply nested JSON document. At the top level, each workspace object contains arrays of:

• datasets — with nested tables (containing columns, measures, partitions), datasourceUsages, and upstreamDataflows
• reports — with datasetId references and createdBy/modifiedBy user identity
• dashboards — with tile metadata and source report references
• dataflows — with datasourceUsages and upstreamDataflows
• datamarts — if applicable
• users — workspace members with their roles
• sensitivityLabel — if MIP labels are applied
• endorsementDetails — certification and promotion status

The richness of this payload depends on the feature flags you pass in the getInfo request body. The key flags are:

{
  "workspaces": ["workspace-guid-1", "workspace-guid-2"],
  "datasetExpressions": true,
  "datasetSchema": true,
  "datasourceDetails": true,
  "getArtifactUsers": true,
  "lineage": true
}

Setting datasetSchema: true is what enables column-level metadata. Setting datasetExpressions: true exposes DAX measure expressions and Power Query M code — powerful for lineage analysis, but sensitive information to handle carefully.

Configuring Tenant Settings and Service Principal Authentication

AAD App Registration

You'll run your scanner as a service principal, not as a user identity. This matters for scheduled execution (no interactive login), auditability (actions appear as the service principal in audit logs), and least-privilege security.

In Azure Active Directory:

1. Register a new application — name it something descriptive like pbi-metadata-scanner-prod
2. Create a client secret (or use a certificate — certificates are preferred in production)
3. Note the Application (client) ID and the tenant ID
4. Do not assign any Graph API permissions — the Scanner API uses Power BI service permissions, not Graph

The service principal needs to be added to a Power BI Security Group that has been explicitly enabled in the Power BI admin portal.

Power BI Tenant Settings to Enable

In the Power BI admin portal, under Tenant Settings, you need to enable three specific settings:

"Allow service principals to use read-only Power BI admin APIs" — This enables the service principal to call admin endpoints at all. Enable it and scope it to your security group.

"Enhance admin APIs responses with detailed metadata" — This is the critical one. Without it, datasetSchema and datasetExpressions flags are silently ignored even if you include them in your request. The API returns success but omits all schema-level detail.

"Enhance admin APIs responses with DAX and mashup expressions" — Required if you want M code and DAX expressions in your payload.

Authentication Code

Here's a reusable authentication module using the Microsoft Authentication Library for Python (msal):

import msal
import os

class PowerBIAuthenticator:
    """
    Handles service principal authentication to the Power BI REST API.
    Uses client credentials flow — no user interaction required.
    """
    
    POWER_BI_RESOURCE = "https://analysis.windows.net/powerbi/api"
    AUTHORITY_BASE = "https://login.microsoftonline.com"
    
    def __init__(self):
        self.tenant_id = os.environ["PBI_TENANT_ID"]
        self.client_id = os.environ["PBI_CLIENT_ID"]
        self.client_secret = os.environ["PBI_CLIENT_SECRET"]
        self._app = None
        self._token_cache = None
    
    def _get_app(self) -> msal.ConfidentialClientApplication:
        if self._app is None:
            self._app = msal.ConfidentialClientApplication(
                client_id=self.client_id,
                client_credential=self.client_secret,
                authority=f"{self.AUTHORITY_BASE}/{self.tenant_id}"
            )
        return self._app
    
    def get_access_token(self) -> str:
        """
        Returns a valid access token, using the MSAL cache to avoid
        unnecessary round-trips to AAD. Tokens are valid for ~60 minutes.
        """
        app = self._get_app()
        
        # Try silent acquisition first (uses MSAL's in-memory cache)
        result = app.acquire_token_silent(
            scopes=[f"{self.POWER_BI_RESOURCE}/.default"],
            account=None
        )
        
        if not result:
            result = app.acquire_token_for_client(
                scopes=[f"{self.POWER_BI_RESOURCE}/.default"]
            )
        
        if "access_token" not in result:
            raise RuntimeError(
                f"Failed to acquire token: {result.get('error_description', 'Unknown error')}"
            )
        
        return result["access_token"]
    
    def get_headers(self) -> dict:
        return {
            "Authorization": f"Bearer {self.get_access_token()}",
            "Content-Type": "application/json"
        }

Never hardcode credentials. The os.environ pattern here works with Azure Key Vault references in App Service, ADF Linked Services, or GitHub Actions secrets — all reasonable places to run this pipeline.

Building the Metadata Harvesting Pipeline

Now let's build the core scanner. We'll structure this as a class that encapsulates the entire scan lifecycle — discovery, batching, initiation, polling, and retrieval.

import requests
import time
import logging
from datetime import datetime, timezone
from typing import Optional
from dataclasses import dataclass, field

logger = logging.getLogger(__name__)

BASE_URL = "https://api.powerbi.com/v1.0/myorg/admin"

@dataclass
class ScanBatchResult:
    scan_id: str
    workspace_ids: list[str]
    status: str = "Pending"
    result: Optional[dict] = None
    error: Optional[str] = None
    initiated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

class TenantScanner:
    """
    Orchestrates a full tenant metadata scan using the Power BI Scanner API.
    Handles batching (100 workspaces per scan), polling, and result aggregation.
    """
    
    BATCH_SIZE = 100
    POLL_INTERVAL_SECONDS = 15
    MAX_POLL_ATTEMPTS = 40  # 10 minutes max wait per batch
    
    def __init__(self, authenticator: PowerBIAuthenticator):
        self.auth = authenticator
        self.session = requests.Session()
    
    def _get(self, endpoint: str, params: dict = None) -> dict:
        """GET with automatic retry on 429."""
        url = f"{BASE_URL}{endpoint}"
        backoff = 2
        
        for attempt in range(5):
            resp = self.session.get(
                url,
                headers=self.auth.get_headers(),
                params=params
            )
            
            if resp.status_code == 200:
                return resp.json()
            elif resp.status_code == 429:
                retry_after = int(resp.headers.get("Retry-After", backoff))
                logger.warning(f"Rate limited. Waiting {retry_after}s before retry.")
                time.sleep(retry_after)
                backoff = min(backoff * 2, 60)
            elif resp.status_code == 401:
                # Token may have expired mid-run; force re-acquisition
                self.session.headers.clear()
                raise RuntimeError("Authentication failed — check service principal configuration")
            else:
                resp.raise_for_status()
        
        raise RuntimeError(f"Failed to GET {endpoint} after 5 attempts")
    
    def _post(self, endpoint: str, body: dict) -> dict:
        """POST with retry on 429."""
        url = f"{BASE_URL}{endpoint}"
        backoff = 2
        
        for attempt in range(5):
            resp = self.session.post(
                url,
                headers=self.auth.get_headers(),
                json=body
            )
            
            if resp.status_code in (200, 202):
                return resp.json()
            elif resp.status_code == 429:
                retry_after = int(resp.headers.get("Retry-After", backoff))
                logger.warning(f"Rate limited on POST. Waiting {retry_after}s.")
                time.sleep(retry_after)
                backoff = min(backoff * 2, 60)
            else:
                resp.raise_for_status()
        
        raise RuntimeError(f"Failed to POST {endpoint} after 5 attempts")
    
    def get_modified_workspace_ids(
        self,
        modified_since: Optional[datetime] = None
    ) -> list[str]:
        """
        Retrieves workspace IDs modified since a given timestamp.
        Pass None for modified_since to get all workspaces (full scan).
        """
        params = {}
        if modified_since:
            # API expects ISO 8601 with Z suffix
            params["modifiedSince"] = modified_since.strftime("%Y-%m-%dT%H:%M:%SZ")
        
        result = self._get("/workspaces/modified", params=params)
        workspace_ids = [ws["id"] for ws in result]
        
        logger.info(f"Discovered {len(workspace_ids)} workspaces to scan")
        return workspace_ids
    
    def initiate_scan(self, workspace_ids: list[str]) -> str:
        """Initiates a scan for up to 100 workspaces. Returns the scanId."""
        if len(workspace_ids) > self.BATCH_SIZE:
            raise ValueError(f"Batch size cannot exceed {self.BATCH_SIZE}")
        
        body = {
            "workspaces": workspace_ids,
            "datasetExpressions": True,
            "datasetSchema": True,
            "datasourceDetails": True,
            "getArtifactUsers": True,
            "lineage": True
        }
        
        result = self._post("/workspaces/getInfo", body=body)
        scan_id = result["id"]
        logger.info(f"Initiated scan {scan_id} for {len(workspace_ids)} workspaces")
        return scan_id
    
    def poll_until_complete(self, scan_id: str) -> str:
        """Polls scan status until Succeeded or Failed. Returns final status."""
        for attempt in range(self.MAX_POLL_ATTEMPTS):
            result = self._get(f"/workspaces/scanStatus/{scan_id}")
            status = result["status"]
            
            if status == "Succeeded":
                logger.info(f"Scan {scan_id} completed successfully")
                return status
            elif status == "Failed":
                error = result.get("error", {})
                raise RuntimeError(
                    f"Scan {scan_id} failed: {error.get('message', 'Unknown error')}"
                )
            
            logger.debug(f"Scan {scan_id} status: {status}. Attempt {attempt + 1}/{self.MAX_POLL_ATTEMPTS}")
            time.sleep(self.POLL_INTERVAL_SECONDS)
        
        raise TimeoutError(f"Scan {scan_id} did not complete within expected time")
    
    def get_scan_result(self, scan_id: str) -> dict:
        """Retrieves the full metadata payload for a completed scan."""
        return self._get(f"/workspaces/scanResult/{scan_id}")
    
    def run_full_scan(
        self,
        modified_since: Optional[datetime] = None
    ) -> list[dict]:
        """
        Runs a complete tenant scan: discovery, batching, all phases.
        Returns a list of workspace metadata objects.
        """
        workspace_ids = self.get_modified_workspace_ids(modified_since)
        
        if not workspace_ids:
            logger.info("No workspaces modified since last scan. Nothing to do.")
            return []
        
        # Split into batches of 100
        batches = [
            workspace_ids[i:i + self.BATCH_SIZE]
            for i in range(0, len(workspace_ids), self.BATCH_SIZE)
        ]
        
        all_workspaces = []
        
        for batch_num, batch in enumerate(batches, 1):
            logger.info(f"Processing batch {batch_num}/{len(batches)}")
            
            scan_id = self.initiate_scan(batch)
            # Respect the 1 scan/second rate limit between initiations
            time.sleep(1.5)
            
            self.poll_until_complete(scan_id)
            result = self.get_scan_result(scan_id)
            
            workspaces = result.get("workspaces", [])
            all_workspaces.extend(workspaces)
            logger.info(
                f"Batch {batch_num} complete: {len(workspaces)} workspaces retrieved"
            )
        
        logger.info(f"Full scan complete. Total workspaces harvested: {len(all_workspaces)}")
        return all_workspaces

This class handles the full lifecycle cleanly. One thing worth noting: we sleep 1.5 seconds between scan initiations rather than 1.0. The Scanner API's rate limit is documented as "1 per second" but in practice the server-side enforcement has some jitter. The extra 500ms prevents spurious 429s without meaningfully slowing down the pipeline.

Designing the Catalog Schema

Raw JSON from the Scanner API is useful but unwieldy for querying. You want to flatten and normalize it into relational tables that enable efficient cross-workspace queries.

Here's a schema designed around the governance questions you'll actually need to answer:

-- Core workspace inventory
CREATE TABLE catalog.workspaces (
    workspace_id        VARCHAR(36) PRIMARY KEY,
    workspace_name      VARCHAR(500) NOT NULL,
    workspace_type      VARCHAR(50),    -- "Workspace", "PersonalGroup", etc.
    state               VARCHAR(50),    -- "Active", "Deleted"
    capacity_id         VARCHAR(36),
    is_on_dedicated_capacity BOOLEAN,
    default_dataset_storage_format VARCHAR(50),
    scanned_at          TIMESTAMP NOT NULL,
    modified_utc        TIMESTAMP
);

-- All datasets across the tenant
CREATE TABLE catalog.datasets (
    dataset_id          VARCHAR(36) PRIMARY KEY,
    workspace_id        VARCHAR(36) NOT NULL REFERENCES catalog.workspaces(workspace_id),
    dataset_name        VARCHAR(500),
    configured_by       VARCHAR(500),   -- The owner's email/UPN
    target_storage_mode VARCHAR(50),    -- "Import", "DirectQuery", "Composite"
    is_refreshable      BOOLEAN,
    created_date        TIMESTAMP,
    content_provider_type VARCHAR(50),
    endorsement         VARCHAR(50),    -- "None", "Promoted", "Certified"
    endorsed_by         VARCHAR(500),
    sensitivity_label_id VARCHAR(100),
    scanned_at          TIMESTAMP NOT NULL
);

-- Tables within datasets
CREATE TABLE catalog.dataset_tables (
    table_id            BIGSERIAL PRIMARY KEY,
    dataset_id          VARCHAR(36) NOT NULL REFERENCES catalog.datasets(dataset_id),
    table_name          VARCHAR(500) NOT NULL,
    table_type          VARCHAR(50),    -- "Table", "CalculatedTable", "LocalDateTable"
    source_expression   TEXT,           -- M/DAX expression if datasetExpressions=true
    row_count           BIGINT,
    scanned_at          TIMESTAMP NOT NULL
);

-- Columns within tables
CREATE TABLE catalog.dataset_columns (
    column_id           BIGSERIAL PRIMARY KEY,
    table_id            BIGINT NOT NULL REFERENCES catalog.dataset_tables(table_id),
    dataset_id          VARCHAR(36) NOT NULL,
    column_name         VARCHAR(500) NOT NULL,
    column_type         VARCHAR(50),    -- "Column", "CalculatedColumn", "Key"
    data_type           VARCHAR(50),
    is_hidden           BOOLEAN,
    description         TEXT,
    sensitivity_label_id VARCHAR(100),
    scanned_at          TIMESTAMP NOT NULL
);

-- Measures within datasets (not tied to specific table in API response)
CREATE TABLE catalog.dataset_measures (
    measure_id          BIGSERIAL PRIMARY KEY,
    dataset_id          VARCHAR(36) NOT NULL REFERENCES catalog.datasets(dataset_id),
    table_name          VARCHAR(500),
    measure_name        VARCHAR(500) NOT NULL,
    expression          TEXT,           -- DAX expression
    is_hidden           BOOLEAN,
    description         TEXT,
    scanned_at          TIMESTAMP NOT NULL
);

-- Data source connections
CREATE TABLE catalog.datasource_connections (
    connection_id       BIGSERIAL PRIMARY KEY,
    dataset_id          VARCHAR(36) REFERENCES catalog.datasets(dataset_id),
    dataflow_id         VARCHAR(36),
    datasource_type     VARCHAR(200),   -- "Sql", "AzureDataLakeStorage", etc.
    connection_details  JSONB,          -- Server, database, path, etc.
    gateway_id          VARCHAR(36),
    datasource_id       VARCHAR(36),
    scanned_at          TIMESTAMP NOT NULL
);

-- Reports
CREATE TABLE catalog.reports (
    report_id           VARCHAR(36) PRIMARY KEY,
    workspace_id        VARCHAR(36) NOT NULL REFERENCES catalog.workspaces(workspace_id),
    report_name         VARCHAR(500),
    report_type         VARCHAR(50),    -- "PowerBIReport", "PaginatedReport"
    dataset_id          VARCHAR(36),
    created_by          VARCHAR(500),
    created_date        TIMESTAMP,
    modified_by         VARCHAR(500),
    modified_date       TIMESTAMP,
    sensitivity_label_id VARCHAR(100),
    scanned_at          TIMESTAMP NOT NULL
);

-- Workspace membership
CREATE TABLE catalog.workspace_users (
    membership_id       BIGSERIAL PRIMARY KEY,
    workspace_id        VARCHAR(36) NOT NULL REFERENCES catalog.workspaces(workspace_id),
    user_identifier     VARCHAR(500),   -- email or group name
    principal_type      VARCHAR(50),    -- "User", "Group", "ServicePrincipal"
    access_right        VARCHAR(50),    -- "Admin", "Member", "Contributor", "Viewer"
    scanned_at          TIMESTAMP NOT NULL
);

-- Lineage: cross-workspace dataset dependencies
CREATE TABLE catalog.lineage_edges (
    edge_id             BIGSERIAL PRIMARY KEY,
    source_workspace_id VARCHAR(36),
    source_dataset_id   VARCHAR(36),
    target_dataset_id   VARCHAR(36),
    target_workspace_id VARCHAR(36),
    edge_type           VARCHAR(50),    -- "DatasetToDataset", "DataflowToDataset"
    scanned_at          TIMESTAMP NOT NULL
);

Notice the JSONB column for connection_details. Datasource connection shapes vary wildly by connector type — a SQL Server connection has server and database fields, while an ADLS Gen2 connection has url. Storing the raw connection object as JSONB lets you query specific fields with PostgreSQL's -> operator while avoiding the nightmare of a sparse table with 40 nullable columns.

Parsing and Loading the Harvest

Now we need to take the raw workspace metadata list and load it into the schema above. The parsing is the most complex part because the JSON nesting is deep and inconsistent — some objects have optional arrays that are simply absent rather than empty.

import pandas as pd
from sqlalchemy import create_engine, text
from datetime import datetime, timezone
from typing import Any

class CatalogLoader:
    """
    Parses raw Scanner API workspace payloads and loads them
    into the normalized catalog schema.
    """
    
    def __init__(self, connection_string: str):
        self.engine = create_engine(connection_string)
        self.scanned_at = datetime.now(timezone.utc)
    
    def _safe_get(self, obj: dict, *keys, default=None) -> Any:
        """Safely traverse nested dicts without KeyError."""
        for key in keys:
            if not isinstance(obj, dict):
                return default
            obj = obj.get(key, default)
        return obj
    
    def parse_workspaces(self, raw_workspaces: list[dict]) -> pd.DataFrame:
        rows = []
        for ws in raw_workspaces:
            rows.append({
                "workspace_id": ws.get("id"),
                "workspace_name": ws.get("name"),
                "workspace_type": ws.get("type"),
                "state": ws.get("state"),
                "capacity_id": ws.get("capacityId"),
                "is_on_dedicated_capacity": ws.get("isOnDedicatedCapacity", False),
                "default_dataset_storage_format": ws.get("defaultDatasetStorageFormat"),
                "scanned_at": self.scanned_at,
                "modified_utc": ws.get("modifiedUtcTime")
            })
        return pd.DataFrame(rows)
    
    def parse_datasets(self, raw_workspaces: list[dict]) -> pd.DataFrame:
        rows = []
        for ws in raw_workspaces:
            workspace_id = ws.get("id")
            for ds in ws.get("datasets", []):
                endorsement_details = ds.get("endorsementDetails", {})
                rows.append({
                    "dataset_id": ds.get("id"),
                    "workspace_id": workspace_id,
                    "dataset_name": ds.get("name"),
                    "configured_by": ds.get("configuredBy"),
                    "target_storage_mode": ds.get("targetStorageMode"),
                    "is_refreshable": ds.get("isRefreshable", False),
                    "created_date": ds.get("createdDate"),
                    "content_provider_type": ds.get("contentProviderType"),
                    "endorsement": self._safe_get(
                        endorsement_details, "endorsement", default="None"
                    ),
                    "endorsed_by": self._safe_get(
                        endorsement_details, "certifiedBy"
                    ),
                    "sensitivity_label_id": self._safe_get(
                        ds, "sensitivityLabel", "labelId"
                    ),
                    "scanned_at": self.scanned_at
                })
        return pd.DataFrame(rows)
    
    def parse_tables_and_columns(
        self, raw_workspaces: list[dict]
    ) -> tuple[pd.DataFrame, pd.DataFrame]:
        table_rows = []
        column_rows = []
        
        for ws in raw_workspaces:
            for ds in ws.get("datasets", []):
                dataset_id = ds.get("id")
                
                for table in ds.get("tables", []):
                    table_name = table.get("name")
                    
                    table_rows.append({
                        "dataset_id": dataset_id,
                        "table_name": table_name,
                        "table_type": table.get("tableType"),
                        "source_expression": table.get("source"),
                        "row_count": table.get("rowCount"),
                        "scanned_at": self.scanned_at
                    })
                    
                    for col in table.get("columns", []):
                        column_rows.append({
                            "dataset_id": dataset_id,
                            "table_name": table_name,
                            "column_name": col.get("name"),
                            "column_type": col.get("columnType"),
                            "data_type": col.get("dataType"),
                            "is_hidden": col.get("isHidden", False),
                            "description": col.get("description"),
                            "sensitivity_label_id": self._safe_get(
                                col, "sensitivityLabel", "labelId"
                            ),
                            "scanned_at": self.scanned_at
                        })
        
        return pd.DataFrame(table_rows), pd.DataFrame(column_rows)
    
    def parse_datasources(self, raw_workspaces: list[dict]) -> pd.DataFrame:
        rows = []
        
        for ws in raw_workspaces:
            # Datasources on datasets
            for ds in ws.get("datasets", []):
                for usage in ds.get("datasourceUsages", []):
                    rows.append({
                        "dataset_id": ds.get("id"),
                        "dataflow_id": None,
                        "datasource_type": self._safe_get(
                            usage, "datasourceInstanceId"
                        ),
                        "connection_details": usage.get("connectionDetails"),
                        "gateway_id": usage.get("gatewayId"),
                        "datasource_id": usage.get("datasourceId"),
                        "scanned_at": self.scanned_at
                    })
            
            # Datasources on dataflows
            for df in ws.get("dataflows", []):
                for usage in df.get("datasourceUsages", []):
                    rows.append({
                        "dataset_id": None,
                        "dataflow_id": df.get("objectId"),
                        "datasource_type": usage.get("datasourceType"),
                        "connection_details": usage.get("connectionDetails"),
                        "gateway_id": usage.get("gatewayId"),
                        "datasource_id": usage.get("datasourceId"),
                        "scanned_at": self.scanned_at
                    })
        
        return pd.DataFrame(rows)
    
    def upsert_dataframe(
        self,
        df: pd.DataFrame,
        table_name: str,
        pk_columns: list[str]
    ):
        """
        Performs an upsert: insert new rows, update existing rows on PK conflict.
        Uses PostgreSQL ON CONFLICT ... DO UPDATE syntax.
        """
        if df.empty:
            return
        
        # Write to a staging table first
        staging = f"{table_name}_staging"
        df.to_sql(
            staging,
            self.engine,
            schema="catalog",
            if_exists="replace",
            index=False
        )
        
        columns = df.columns.tolist()
        pk_str = ", ".join(pk_columns)
        update_cols = [c for c in columns if c not in pk_columns]
        update_str = ", ".join([f"{c} = EXCLUDED.{c}" for c in update_cols])
        
        upsert_sql = f"""
            INSERT INTO catalog.{table_name}
            SELECT * FROM catalog.{staging}
            ON CONFLICT ({pk_str})
            DO UPDATE SET {update_str}
        """
        
        with self.engine.connect() as conn:
            conn.execute(text(upsert_sql))
            conn.execute(text(f"DROP TABLE catalog.{staging}"))
            conn.commit()
    
    def load_all(self, raw_workspaces: list[dict]):
        """Full ETL: parse all entities and upsert into catalog."""
        logger.info("Loading workspace metadata...")
        
        ws_df = self.parse_workspaces(raw_workspaces)
        self.upsert_dataframe(ws_df, "workspaces", ["workspace_id"])
        
        ds_df = self.parse_datasets(raw_workspaces)
        self.upsert_dataframe(ds_df, "datasets", ["dataset_id"])
        
        table_df, col_df = self.parse_tables_and_columns(raw_workspaces)
        # Tables don't have a stable GUID — use dataset_id + table_name as PK
        # The BIGSERIAL PK in the schema uses the composite for upsert logic
        # In practice: truncate-and-reload tables/columns per dataset, not global upsert
        self._reload_table_data(table_df, col_df, ds_df["dataset_id"].tolist())
        
        src_df = self.parse_datasources(raw_workspaces)
        logger.info(
            f"Loaded: {len(ws_df)} workspaces, {len(ds_df)} datasets, "
            f"{len(table_df)} tables, {len(col_df)} columns, "
            f"{len(src_df)} datasource connections"
        )
    
    def _reload_table_data(
        self,
        table_df: pd.DataFrame,
        col_df: pd.DataFrame,
        dataset_ids: list[str]
    ):
        """
        For tables and columns, we do a delete-and-replace per dataset
        rather than row-level upsert, because the BIGSERIAL PKs are
        not stable across scans.
        """
        if not dataset_ids:
            return
        
        id_list = ", ".join([f"'{d}'" for d in dataset_ids])
        
        with self.engine.connect() as conn:
            conn.execute(text(
                f"DELETE FROM catalog.dataset_columns WHERE dataset_id IN ({id_list})"
            ))
            conn.execute(text(
                f"DELETE FROM catalog.dataset_tables WHERE dataset_id IN ({id_list})"
            ))
            conn.commit()
        
        if not table_df.empty:
            table_df.to_sql(
                "dataset_tables", self.engine, schema="catalog",
                if_exists="append", index=False
            )
        
        if not col_df.empty:
            col_df.to_sql(
                "dataset_columns", self.engine, schema="catalog",
                if_exists="append", index=False
            )

The delete-and-replace pattern for tables and columns deserves explanation. The Scanner API doesn't return stable GUIDs for table and column objects — only datasets, reports, workspaces, and dataflows have stable GUIDs. Since we're using BIGSERIAL surrogate keys for tables and columns, a row-level upsert would require matching on composite natural keys, which works but is slower. For tables and columns, deleting all rows for the affected datasets and reinserting is simpler, faster, and equally correct.

Building the Governance Reporting Layer

The catalog database is now populated. Now we build the Power BI report that turns this data into governance intelligence.

Key Governance Queries

Before building visuals, get your SQL right. These are the queries that actually answer the governance questions your stakeholders ask.

All datasets connecting to a specific source system (critical for decommissioning planning):

SELECT DISTINCT
    w.workspace_name,
    d.dataset_name,
    d.configured_by,
    d.endorsement,
    dc.datasource_type,
    dc.connection_details->>'server' AS server,
    dc.connection_details->>'database' AS database_name
FROM catalog.datasource_connections dc
JOIN catalog.datasets d ON dc.dataset_id = d.dataset_id
JOIN catalog.workspaces w ON d.workspace_id = w.workspace_id
WHERE 
    dc.connection_details->>'server' ILIKE '%legacy-erp-prod%'
    OR dc.connection_details->>'database' ILIKE '%AdventureWorks%'
ORDER BY w.workspace_name, d.dataset_name;

Column-level PII inventory (for data privacy audits):

SELECT
    w.workspace_name,
    d.dataset_name,
    dt.table_name,
    dc.column_name,
    dc.data_type,
    dc.sensitivity_label_id,
    d.endorsement
FROM catalog.dataset_columns dc
JOIN catalog.dataset_tables dt ON dc.table_id = dt.table_id
JOIN catalog.datasets d ON dt.dataset_id = d.dataset_id
JOIN catalog.workspaces w ON d.workspace_id = w.workspace_id
WHERE dc.sensitivity_label_id IS NOT NULL
ORDER BY w.workspace_name, d.dataset_name, dt.table_name, dc.column_name;

Orphaned datasets (datasets with no reports built on them — candidates for cleanup):

SELECT
    w.workspace_name,
    d.dataset_id,
    d.dataset_name,
    d.configured_by,
    d.created_date,
    d.is_refreshable
FROM catalog.datasets d
JOIN catalog.workspaces w ON d.workspace_id = w.workspace_id
LEFT JOIN catalog.reports r ON d.dataset_id = r.dataset_id
WHERE r.report_id IS NULL
ORDER BY d.created_date DESC;

Workspace coverage by capacity (for capacity planning):

SELECT
    w.capacity_id,
    w.is_on_dedicated_capacity,
    COUNT(DISTINCT w.workspace_id) AS workspace_count,
    COUNT(DISTINCT d.dataset_id) AS dataset_count,
    COUNT(DISTINCT r.report_id) AS report_count
FROM catalog.workspaces w
LEFT JOIN catalog.datasets d ON w.workspace_id = d.workspace_id
LEFT JOIN catalog.reports r ON w.workspace_id = r.workspace_id
GROUP BY w.capacity_id, w.is_on_dedicated_capacity
ORDER BY workspace_count DESC;

These queries are your starting point. In Power BI Desktop, connect to your catalog database with DirectQuery rather than Import — the catalog data is already structured for fast query performance, and you want governance reports to always show current data without a separate refresh cycle.

Governance Report Structure

Build your governance report with three pages:

Page 1 — Asset Inventory. A matrix showing workspace → dataset → report counts, filtered by capacity, endorsement level, and sensitivity label. A treemap showing dataset distribution by workspace is excellent here for spotting "workspace sprawl." The endorsement filter is particularly useful — you can immediately see how many of your datasets are certified versus unendorsed. This connects well with the certification workflows described in Implementing Power BI Dataset Certification and Endorsement Workflows to Establish a Trusted Enterprise Data Catalog.

Page 2 — Data Source Lineage. A table showing every unique (datasource_type, server, database) combination with the count of datasets using it. Clicking a row cross-filters to show the specific datasets and their workspaces. This is the "what breaks if we retire this server" page.

Page 3 — Governance Compliance. A set of KPI cards: percentage of datasets with endorsement, percentage of sensitive-labeled assets, count of orphaned datasets, count of workspaces without a dedicated capacity admin. Below the cards, a detail table with drill-through capability to individual workspace/dataset records.

Incremental Scheduling and Operational Considerations

Scheduling the Scan

Run your scanner on a schedule that balances freshness against API consumption. A reasonable cadence for most enterprises:

• Full scan: Weekly (Sunday night), captures workspaces that may have been deleted or renamed
• Incremental scan: Every 4–6 hours, uses modifiedSince to pick up new and changed assets
• Lineage-only scan: Daily, since lineage changes less frequently than schema

Store the timestamp of your last successful scan in a small control table:

CREATE TABLE catalog.scan_control (
    scan_type       VARCHAR(50) PRIMARY KEY,
    last_scan_utc   TIMESTAMP NOT NULL,
    workspace_count INTEGER,
    status          VARCHAR(20)
);

At the start of each incremental run, read last_scan_utc for scan_type = 'incremental'. After a successful run, update it. If the run fails, leave it unchanged so the next run covers the missed window.

Detecting Deleted Assets

The modifiedSince endpoint returns workspaces that were modified, not workspaces that were deleted. Deleted workspaces disappear from the results. To detect deletions, maintain a separate process:

def identify_deleted_workspaces(
    all_current_ids: list[str],
    engine
) -> list[str]:
    """
    Compares workspace IDs from the current scan against the catalog.
    Returns IDs present in the catalog but absent from the current scan.
    Only call this during a full scan — incremental scans don't return all IDs.
    """
    with engine.connect() as conn:
        result = conn.execute(
            text("SELECT workspace_id FROM catalog.workspaces WHERE state = 'Active'")
        )
        catalog_ids = {row[0] for row in result}
    
    current_set = set(all_current_ids)
    deleted = catalog_ids - current_set
    
    if deleted:
        with engine.connect() as conn:
            id_list = ", ".join([f"'{d}'" for d in deleted])
            conn.execute(text(
                f"""UPDATE catalog.workspaces 
                    SET state = 'Deleted', scanned_at = NOW() 
                    WHERE workspace_id IN ({id_list})"""
            ))
            conn.commit()
    
    return list(deleted)

This soft-delete pattern — marking workspaces as state = 'Deleted' rather than physically removing them — is important for audit trails. Your governance report should show a "recently deleted workspaces" panel so admins can verify intentional versus accidental deletions. This pairs naturally with the audit logging patterns covered in Power BI Governance: Workspaces, Permissions, and Audit Logging.

Handling Large Tenants

For tenants with more than 1,000 workspaces, sequential batch processing becomes slow. You can parallelize the scan initiations using concurrent futures, but you have to respect the rate limits carefully:

from concurrent.futures import ThreadPoolExecutor, as_completed

def run_parallel_scans(
    scanner: TenantScanner,
    workspace_id_batches: list[list[str]],
    max_workers: int = 3
) -> list[dict]:
    """
    Initiates multiple scans in parallel, then polls all concurrently.
    max_workers=3 is conservative — don't exceed 5 to avoid throttling.
    """
    all_workspaces = []
    
    def process_batch(batch: list[str]) -> list[dict]:
        scan_id = scanner.initiate_scan(batch)
        time.sleep(1.5)  # Still respect per-request rate limit
        scanner.poll_until_complete(scan_id)
        result = scanner.get_scan_result(scan_id)
        return result.get("workspaces", [])
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(process_batch, batch): batch
            for batch in workspace_id_batches
        }
        
        for future in as_completed(futures):
            try:
                workspaces = future.result()
                all_workspaces.extend(workspaces)
            except Exception as e:
                batch = futures[future]
                logger.error(f"Batch failed: {e}. Batch size: {len(batch)}")
                # Don't re-raise — log and continue with other batches
    
    return all_workspaces

With max_workers=3 and batches of 100 workspaces, you can scan 1,500 workspaces in roughly 3–4 minutes rather than 15–20. That's the practical ceiling before you start seeing consistent 429 responses.

Integrating with the Broader Governance Stack

The metadata catalog you've built doesn't exist in isolation. It should feed into — and draw from — other governance systems.

Activity logs integration: The Scanner API tells you what exists, but the activity logs tell you what gets used. Join catalog.reports with activity log view counts on report_id to identify orphaned or underused assets.

Sensitivity label cross-reference: The Scanner API returns sensitivity_label_id as a GUID, not a human-readable label name. You need to maintain a lookup table mapping label GUIDs to label names, which you can pull from the Microsoft Information Protection API or hardcode from your tenant's label configuration. This connects directly to the sensitivity label management patterns in Implementing Power BI Tenant Settings and Sensitivity Labels for Enterprise Data Protection and Compliance.

Dataset endorsement governance loop: Your catalog report can surface unendorsed datasets in high-traffic workspaces, and you can build a Power Automate flow that notifies workspace admins to endorse or certify their assets. Certification status flows back into the catalog at the next scan.

Cross-workspace lineage: The Scanner API's lineage flag populates upstreamDataflows and upstreamDatasets arrays that describe which datasets depend on which dataflows. Combined with the patterns in Implementing Power BI Workspace-Level Lineage and Impact Analysis to Manage Dataset Dependencies Across the Enterprise, you can build a full dependency graph that answers "if I change this dataflow schema, which downstream reports break?" — a question that's nearly impossible to answer manually at enterprise scale.

Hands-On Exercise

Build a complete scanner pipeline for your own tenant with these specific deliverables:

Step 1: Register a service principal and enable the three required tenant settings. Verify your setup by calling GET /v1.0/myorg/admin/workspaces/modified and confirming you get a workspace list.

Step 2: Run the TenantScanner.run_full_scan() method and save the raw JSON output to a file. Open it and manually inspect the structure for three workspaces — identify how tables, columns, datasource usages, and sensitivity labels are nested.

Step 3: Create the catalog schema in a PostgreSQL database (local or cloud). Run CatalogLoader.load_all() against your raw JSON. Write and execute the "orphaned datasets" query and the "datasource inventory" query. How many datasets in your tenant have no reports?

Step 4: Build a governance report in Power BI Desktop connecting via DirectQuery to your catalog database. Include at minimum: a workspace inventory table, a datasource connection breakdown by type, and an endorsement coverage KPI. Publish it to a dedicated "BI Governance" workspace.

Step 5: Schedule the incremental scan to run every 6 hours using either Azure Functions (HTTP trigger + Timer trigger), an Azure Data Factory pipeline, or a simple cron job on a server. Confirm the scan_control table updates correctly after each run.

The expected outcome: you should be able to answer "which datasets connect to our SQL Server named PROD-DW-01?" in under 10 seconds by querying your catalog, rather than clicking through 200 workspaces in the admin portal.

Common Mistakes & Troubleshooting

"My scan results don't include column details even though I passed datasetSchema: true." This is almost always a missing tenant setting. Go to the admin portal and verify "Enhance admin APIs responses with detailed metadata" is enabled and includes your service principal's security group. The API returns HTTP 200 with partial data when this setting is disabled — it doesn't return an error, which makes it easy to miss.

"My service principal gets 403 on the getInfo endpoint but 200 on other admin endpoints." The Scanner API specifically requires the "Allow service principals to use read-only Power BI admin APIs" tenant setting. Being a Fabric Admin is not sufficient on its own for service principals — human admins bypass this setting, service principals don't.

"Scan status stays in 'Running' forever." This occasionally happens for workspaces containing very large datasets (multi-billion row models) or datasets with complex Power Query expressions. After MAX_POLL_ATTEMPTS is reached, log the failed scan and move on — your incremental scan will retry those workspaces at the next cycle. Never block your entire pipeline on a single stuck scan.

"The datasource_type field is null for some datasource usages." The Scanner API returns datasource type differently depending on whether the datasource has been used at least once. Datasources in datasets that have never been refreshed may have incomplete connection metadata. This is a known behavior — treat nulls in datasource_type as "unresolved" and surface them in your governance report as items needing attention.

"Column-level data is there for some datasets but missing for others." Import-mode datasets with Direct Lake storage on Fabric may return column metadata differently from classic import datasets. Datasets using DirectQuery or composite models may return fewer schema details, particularly for the DirectQuery portions. Additionally, personal workspace datasets (workspaces of type PersonalGroup) often return limited metadata even with all flags enabled.

"My upsert logic is slow for large catalogs." The staging table upsert pattern works fine up to about 100,000 rows. Beyond that, add indexes on your staging tables before the INSERT ... ON CONFLICT operation. Also consider batching the upsert by workspace rather than doing one massive operation — smaller transactions are easier to retry on deadlock.

Summary & Next Steps

You've built a complete metadata harvesting system: an authenticated scanner that traverses your entire tenant, a normalized catalog schema that stores workspace, dataset, table, column, datasource, and lineage data, a loading pipeline with upsert semantics and incremental update logic, and a governance reporting layer that turns raw metadata into actionable intelligence.

The Scanner API is one of the most powerful — and most underutilized — capabilities in the Power BI admin toolbox. Most organizations that struggle with "data catalog sprawl" or "we don't know what we have" problems are sitting on a complete solution they just haven't wired up yet. What you've built here is the foundation of a proper enterprise data catalog built entirely on native Power BI capabilities, without requiring a third-party catalog tool.

Where to go from here:

• Extend the catalog to include dataflow refresh history by joining Scanner API results with the activity log data
• Add automated compliance scoring: build a DAX measure that scores each workspace on a governance rubric (has certified datasets, has a sensitivity label, has a designated admin, has no orphaned reports)
• Implement change detection: compare two scan snapshots to generate a "what changed" report, invaluable for tracking unauthorized schema changes or unexpected workspace deletions
• Layer the catalog into a broader Azure governance architecture by shipping your catalog tables to Azure Purview (now Microsoft Purview Data Catalog), which has native connectors for Power BI metadata and can correlate it with your SQL, Azure Data Lake, and other source system lineage — connecting the picture described in Power BI and Azure Synapse: Building Enterprise Analytics Architecture from Scratch
• Pair the schema-level metadata with XMLA endpoint access — covered in Implementing Power BI XMLA Endpoint for Advanced Dataset Management and Third-Party Tool Integration — to build automated documentation generators that produce data dictionaries directly from your semantic models

The organization that knows its analytics estate completely is the one that can govern it, scale it, and trust it. That's what you've started building today.