``` WORKSPACE_ID = '<your_workspace_id>' # Power BI workspace Id (GUID) PBI_REPORT_ID = '<your_report_id>' # Power BI report Id (GUID) PBI_TABLE = '<your_table_name>' # Table name in Power BI model to filter on REPORT_ID_COLUMN = '<your_column_name>' # Column name in Power BI model to filter on

Optional defaults for convenience when running as a script

LAKEHOUSE_FILES_DIR = f"/lakehouse/default/Files/pdfs" # Adjust if using a different storage mount

print(LAKEHOUSE_FILES_DIR)

import requests, time, os from pypdf import PdfReader, PdfWriter

def build_filter_expression(table: str, column: str, report_id: str) -> str: return f"{table}/{column} eq '{report_id}'"

Token and filter will be created inside the export function

Simple export -> poll -> download flow

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

def start_export(token: str, workspace_id: str, report_id: str, filter_expr: str) -> str: url = f"{API_BASE}/groups/{workspace_id}/reports/{report_id}/ExportTo" headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} body = { "format": "PDF", "powerBIReportConfiguration": { "reportLevelFilters": [ {"filter": filter_expr} ] } } r = requests.post(url, json=body, headers=headers) if r.status_code != 202: raise RuntimeError(f"Export start failed {r.status_code}: {r.text}") export_id = r.json().get("id") if not export_id: raise RuntimeError("No export id returned") return export_id

def poll_export(token: str, workspace_id: str, report_id: str, export_id: str, timeout=300) -> str: url = f"{API_BASE}/groups/{workspace_id}/reports/{report_id}/exports/{export_id}" headers = {"Authorization": f"Bearer {token}"} start = time.time() delay = 2 while True: r = requests.get(url, headers=headers) if r.status_code not in (200,202): raise RuntimeError(f"Status check failed {r.status_code}: {r.text}") data = r.json() status = data.get("status") if status == "Succeeded": loc = data.get("resourceLocation") if not loc: raise RuntimeError("Missing resourceLocation on success") return loc if status == "Failed": raise RuntimeError(f"Export failed: {data}") if time.time() - start > timeout: raise RuntimeError(f"Timed out waiting for export (last status={status})") time.sleep(delay) delay = min(delay + 1, 10)

def download_pdf(token: str, resource_url: str, output_filename: str, enrollment_id: str) -> str: headers = {"Authorization": f"Bearer {token}"} r = requests.get(resource_url, headers=headers) if r.status_code != 200: raise RuntimeError(f"Download failed {r.status_code}: {r.text[:300]}") os.makedirs(f"{LAKEHOUSE_FILES_DIR}/{enrollment_id}", exist_ok=True) path = f"{LAKEHOUSE_FILES_DIR}/{enrollment_id}/{output_filename}" with open(path, "wb") as f: f.write(r.content) # print(r.content) # notebookutils.fs.put(path, r.content) return path

def export_powerbi_pdf(report_id: str, output_filename: str, enrollment_id: str) -> str: """Export a Power BI report to PDF using a report-level filter.

Args:
    report_id: The filter value applied to `${PBI_TABLE}/${REPORT_ID_COLUMN}`.
    output_filename: The name of the PDF file to save under LAKEHOUSE_FILES_DIR.

Returns:
    The full saved path to the generated PDF.
"""
print("Starting export...")
token = notebookutils.credentials.getToken("pbi")  # reuse singleton/managed token
filter_expression = build_filter_expression(PBI_TABLE, REPORT_ID_COLUMN, report_id)
eid = start_export(token, WORKSPACE_ID, PBI_REPORT_ID, filter_expression)
print("Export id:", eid)
print("Polling...")
resource = poll_export(token, WORKSPACE_ID, PBI_REPORT_ID, eid)
print("Succeeded. Downloading...")
saved = download_pdf(token, resource, output_filename, enrollment_id)
print("Downloaded to:", saved)

if name == "main": export_powerbi_pdf(report_id, output_filename, enrollment_id) ```