Powerbi - List users of semantic models and reports

app.py

# COMPLETE WORKSPACE USER CATALOG - Azure Default Credentials
# Gets all workspaces and their users, with rate limiting and filtering
# API endpoint supports 200 request per hour..

import time
import json
import requests
from azure.identity import DefaultAzureCredential

# ========= CONFIGURATION =========
SCOPE = "https://analysis.windows.net/powerbi/api/.default"

PBI_BASE = "https://api.powerbi.com/v1.0/myorg"
ADMIN_GROUPS = f"{PBI_BASE}/admin/groups?$top=5" 
ADMIN_GROUP_BYID = lambda gid: f"{PBI_BASE}/admin/groups/{gid}?$expand=users,datasets,reports,dashboards"

# ========= OPTIONAL FILTERS =========
FILTER_WORKSPACE_IDS = []
FILTER_WORKSPACE_NAMES = []

# ========= AUTHENTICATION =========
credential = DefaultAzureCredential()

def get_token():
    """Get Power BI access token using Azure Default Credentials"""
    token = credential.get_token(SCOPE)
    return token.token

# ========= API UTILITIES =========
def pbi_get(url, token, max_retries=5):
    """Make GET request with rate limiting retry logic"""
    for attempt in range(max_retries):
        resp = requests.get(url, headers={"Authorization": f"Bearer {token}"})
        
        if resp.status_code == 429:
            retry_after = int(resp.headers.get("Retry-After", "5"))
            wait_time = retry_after + attempt
            print(f"  ⚠️ Rate limited (429). Waiting {wait_time}s before retry...")
            time.sleep(wait_time)
            continue
        
        if resp.status_code >= 400:
            raise RuntimeError(f"GET {url} failed: {resp.status_code} {resp.text}")
        
        return resp.json()
    
    raise RuntimeError("Exceeded max retries due to throttling.")

def enumerate_admin_groups(token):
    """Page through admin/groups to list all workspaces"""
    groups = []
    url = ADMIN_GROUPS
    page = 1
    
    while True:
        print(f"  Fetching workspace page {page}...")
        data = pbi_get(url, token)
        page_groups = data.get("value", [])
        groups.extend(page_groups)
        print(f"    Found {len(page_groups)} workspaces (total: {len(groups)})")
        
        next_link = data.get("@odata.nextLink")
        if not next_link:
            break
        url = next_link
        page += 1
    
    return groups

def filter_workspaces(workspaces):
    """Filter workspaces by configured IDs and/or names"""
    if not FILTER_WORKSPACE_IDS and not FILTER_WORKSPACE_NAMES:
        return workspaces
    
    id_allow = {wid.lower() for wid in FILTER_WORKSPACE_IDS}
    name_allow = {name.lower() for name in FILTER_WORKSPACE_NAMES}
    
    filtered = []
    for ws in workspaces:
        wid = (ws.get("id") or "").lower()
        name = (ws.get("name") or "").lower()
        
        if id_allow and wid not in id_allow:
            continue
        if name_allow and name not in name_allow:
            continue
        
        filtered.append(ws)
    
    print(f"\n  Filtered from {len(workspaces)} to {len(filtered)} workspaces")
    return filtered

def expand_workspace_users(group_id, group_name, token):
    """Get workspace users and catalog of artifacts"""
    data = pbi_get(ADMIN_GROUP_BYID(group_id), token)
    all_entries = []
    
    # Workspace-level users (primary access control)
    for u in data.get("users", []):
        all_entries.append({
            "workspaceId": group_id,
            "workspaceName": group_name,
            "principalType": u.get("principalType"),
            "identifier": u.get("identifier"),
            "emailAddress": u.get("emailAddress"),
            "displayName": u.get("displayName"),
            "workspaceAccessRight": u.get("groupUserAccessRight"),
        })
    
    datasets = data.get("datasets", [])
    reports = data.get("reports", [])
    dashboards = data.get("dashboards", [])
    
    return {
        "users": all_entries,
        "datasets": [{"id": ds.get("id"), "name": ds.get("name")} for ds in datasets],
        "reports": [{"id": r.get("id"), "name": r.get("name")} for r in reports],
        "dashboards": [{"id": d.get("id"), "name": d.get("displayName")} for d in dashboards]
    }

# ========= MAIN EXECUTION =========
print("="*70)
print("POWER BI WORKSPACE USER CATALOG")
print("="*70)

# Step 1: Authenticate
print("\n1️⃣ Authenticating...")
try:
    token = get_token()
    print("   ✓ Authentication successful")
except Exception as e:
    print(f"   ✗ Authentication failed: {e}")
    raise

# Step 2: Get all workspaces
print("\n2️⃣ Enumerating workspaces...")
workspaces = enumerate_admin_groups(token)
print(f"   ✓ Found {len(workspaces)} total workspaces")

# Step 3: Apply filters
if FILTER_WORKSPACE_IDS or FILTER_WORKSPACE_NAMES:
    print("\n3️⃣ Applying filters...")
    workspaces = filter_workspaces(workspaces)
    if not workspaces:
        print("   ⚠️ No workspaces matched filters. Stopping.")
        raise SystemExit
else:
    print("\n3️⃣ No filters configured - processing all workspaces")

# Step 4: Expand users for each workspace
print(f"\n4️⃣ Expanding users for {len(workspaces)} workspaces...")
catalog = []
workspace_details = []

for i, ws in enumerate(workspaces, 1):
    ws_id = ws["id"]
    ws_name = ws.get("name", "Unnamed")
    
    print(f"   [{i}/{len(workspaces)}] {ws_name}...", end=" ")
    
    try:
        result = expand_workspace_users(ws_id, ws_name, token)
        users = result["users"]
        catalog.extend(users)
        
        # Create minimal user objects without workspace redundancy
        minimal_users = [
            {
                "principalType": u["principalType"],
                "identifier": u["identifier"],
                "emailAddress": u["emailAddress"],
                "displayName": u["displayName"],
                "workspaceAccessRight": u["workspaceAccessRight"]
            }
            for u in users
        ]
        
        workspace_details.append({
            "workspaceId": ws_id,
            "workspaceName": ws_name,
            "userCount": len(users),
            "datasetCount": len(result["datasets"]),
            "reportCount": len(result["reports"]),
            "dashboardCount": len(result["dashboards"]),
            "users": minimal_users,
            "datasets": result["datasets"],
            "reports": result["reports"],
            "dashboards": result["dashboards"]
        })
        
        print(f"✓ {len(users)} users")
    except Exception as e:
        print(f"✗ Error: {e}")
        continue

# Step 5: Save results
print(f"\n5️⃣ Saving results...")
users_file = "workspace_users.json"
details_file = "workspace_details.json"

with open(users_file, "w", encoding="utf-8") as f:
    json.dump(catalog, f, indent=2, ensure_ascii=False)

with open(details_file, "w", encoding="utf-8") as f:
    json.dump(workspace_details, f, indent=2, ensure_ascii=False)

print(f"   ✓ Saved {len(catalog)} user entries to {users_file}")
print(f"   ✓ Saved workspace details to {details_file}")

# Summary
print("\n" + "="*70)
print("SUMMARY")
print("="*70)
print(f"Workspaces processed: {len(workspaces)}")
print(f"Total user entries:   {len(catalog)}")
print(f"Output files:")
print(f"  - {users_file} (user permissions)")
print(f"  - {details_file} (workspace artifacts catalog)")
print("="*70)

# Display sample data
if catalog:
    print("\n📋 Sample user entries:")
    for entry in catalog[:5]:
        print(f"  • {entry.get('displayName')} ({entry.get('emailAddress')})")
        print(f"    Workspace: {entry.get('workspaceName')}")
        print(f"    Access: {entry.get('workspaceAccessRight')}")
        print()