Script to change dataset for a Power BI report

power_bi_report_change_dataset.py

import requests
import adal
import json

# Parameters:
GROUP_ID = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
# AAD Client ID -- To get this, go here: https://dev.powerbi.com/apps
CLIENT_ID = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"

AUTHENTICATION_URL = 'https://login.microsoftonline.com/egmont.onmicrosoft.com'
RESOURCE_URL='https://analysis.windows.net/powerbi/api'
datasets_url = f'https://api.powerbi.com/v1.0/myorg/groups/{GROUP_ID}/datasets'

context = adal.AuthenticationContext(
    authority=AUTHENTICATION_URL,
    validate_authority=True,
    api_version=None)

token = context.acquire_token_with_username_password(
    resource=RESOURCE_URL,
    username=input('Username > '),
    password=input('Password > '),
    client_id=CLIENT_ID)

res = requests.get(
    datasets_url,
    headers={'Authorization': 'Bearer ' + token['accessToken']})
datasets = res.json()['value']

reports_url = f'https://api.powerbi.com/v1.0/myorg/groups/{GROUP_ID}/reports'
res = requests.get(reports_url, headers={'Authorization': 'Bearer ' + token['accessToken']})
reports = res.json()['value']

# Create list of dicts with names and ids for reports and datasets
datasets_and_reports = []
for ds in datasets:
    if not 'BI Core' in ds['name']:
        continue
    print (ds['name'])
    for r in reports:
        if ds['id'] == r['datasetId']:
            print(f'\t\t\t\t{r["name"]}')
            datasets_and_reports.append({'dsname': ds['name'], 'dsid': ds['id'], 'rname': r['name'], 'rid': r['id']})

# Filter the list to only contain reports, based on core models, and retain only the version number of the dataset name.
core_model_versions = [(ds['name'], ds['id'], ds['name'].strip('BI Core').strip()) for ds in datasets if 'BI Core' in ds['name']]
# Filter datasets_and_reports so it does not contain reports that were uploaded as part of a dataset.
only_custom_reports = [_ for _ in datasets_and_reports if 'BI Core' in _['dsname'] and not 'BI Core' in _['rname']]

sorted_core_model_versions = sorted(core_model_versions, key=lambda model: tuple(model[2].strip().split('.')), reverse=True)

for i, x in enumerate(only_custom_reports):
    print (f'{i}\t{x["rname"]}')

target_report = None
target_dataset = None

report_no = input('\nUpgrade report? (type number from list above)> ')
if report_no.isdigit():
    report_no = int(report_no)
    if len(only_custom_reports) <= report_no:
        ValueError('Not a report. Exiting.')
    target_report = only_custom_reports[report_no]
else:
    raise ValueError('This requires a number. Exiting.')

for i, model in enumerate(sorted_core_model_versions):
    model_name, model_id, model_version = model
    print(f'{i}\t{model_name}')
while True:
    target_dataset = input('Choose dataset to bind to > ')
    if target_dataset.isdigit():
        if int(target_dataset) < len(sorted_core_model_versions):
            target_dataset = sorted_core_model_versions[int(target_dataset)]
            break
    print('Faulty input... try again.')

print(f'\n\tRebind "{target_report["rname"]}" to "{target_dataset[0]}"?')

if input('\t(y/n) >').lower() == 'y':
    target_dataset_id = target_dataset[1]
    report_id = target_report['rid']
    rebind_url = f'https://api.powerbi.com/v1.0/myorg/groups/{GROUP_ID}/reports/{report_id}/Rebind'
    payload = json.dumps({"datasetId": target_dataset_id})
    headers = {
        'Content-Type': 'application/json',
        'Authorization': 'Bearer ' + token['accessToken']
    }
    res = requests.post(rebind_url, headers=headers, data=payload)
    if res.status_code == 200:
        print('Done.')
    else:
        print('Try again')
else:
    exit('Do nothing. Exit.')