# app.py
# Import necessary modules
from services.pbiembedservice import PbiEmbedService
from config import BaseConfig # Import your configuration class
import streamlit as st
# Define main function
def main():
# Initialize Streamlit
st.set_page_config(page_title="Economist Impact")
# Load configuration
config = BaseConfig() # Instantiate your configuration class
# Render static HTML page or welcome message
st.markdown("## Welcome to your Power BI Report")
# Handle embedding info request
if st.button('Get Embed Info'):
try:
embed_service = PbiEmbedService(config) # Pass config to PbiEmbedService constructor if needed
embed_info = embed_service.get_embed_params_for_single_report(config.WORKSPACE_ID, config.REPORT_ID)
st.json(embed_info)
except Exception as ex:
st.error(str(ex))
# Entry point of the Streamlit application
if __name__ == '__main__':
main()
# services/aadservice.py
from config import BaseConfig
import msal
# Get Token
class AadService:
def __init__(self):
self.config = BaseConfig()
def get_access_token(self):
try:
if self.config.AUTHENTICATION_MODE.lower() == 'serviceprincipal':
# Correcting the authority URL for the service principal
authority = self.config.AUTHORITY_URL.replace('organizations', self.config.TENANT_ID)
clientapp = msal.ConfidentialClientApplication(
client_id=self.config.CLIENT_ID,
client_credential=self.config.CLIENT_SECRET,
authority=authority
)
response = clientapp.acquire_token_for_client(scopes=self.config.SCOPE_BASE)
elif self.config.AUTHENTICATION_MODE.lower() == 'usernamepassword':
clientapp = msal.PublicClientApplication(
self.config.CLIENT_ID,
authority=self.config.AUTHORITY_URL
)
accounts = clientapp.get_accounts(username=self.config.POWER_BI_USER)
if accounts:
response = clientapp.acquire_token_silent(self.config.SCOPE_BASE, account=accounts[0])
if not response:
response = clientapp.acquire_token_by_username_password(
self.config.POWER_BI_USER,
self.config.POWER_BI_PASS,
scopes=self.config.SCOPE_BASE
)
try:
return response['access_token']
except KeyError:
raise Exception(response.get('error_description', 'Unknown error occurred'))
except Exception as ex:
raise Exception('Error retrieving Access token\n' + str(ex))
# services/pbiembedservice.py
from services.aadservice import AadService
from models.reportconfig import ReportConfig
from models.embedtoken import EmbedToken
from models.embedconfig import EmbedConfig
from models.embedtokenrequestbody import EmbedTokenRequestBody
import requests
import json
class PbiEmbedService:
def __init__(self, config):
self.config = config
self.aad_service = AadService() # Initialize AadService without passing any arguments
def get_embed_params_for_single_report(self, workspace_id, report_id, additional_dataset_id=None):
'''Get embed params for a report and a workspace
Args:
workspace_id (str): Workspace Id
report_id (str): Report Id
additional_dataset_id (str, optional): Dataset Id different than the one bound to the report. Defaults to None.
Returns:
str: JSON representation of EmbedConfig
'''
report_url = f'https://api.powerbi.com/v1.0/myorg/groups/{workspace_id}/reports/{report_id}'
api_response = requests.get(report_url, headers=self.get_request_header())
if api_response.status_code != 200:
raise Exception(f'Error while retrieving Embed URL\n{api_response.reason}:\t{api_response.text}\nRequestId:\t{api_response.headers.get("RequestId")}')
api_response = json.loads(api_response.text)
report = ReportConfig(api_response['id'], api_response['name'], api_response['embedUrl'])
dataset_ids = [api_response['datasetId']]
# Append additional dataset to the list to achieve dynamic binding later
if additional_dataset_id is not None:
dataset_ids.append(additional_dataset_id)
embed_token = self.get_embed_token_for_single_report_single_workspace(report_id, dataset_ids, workspace_id)
embed_config = EmbedConfig(embed_token.tokenId, embed_token.token, embed_token.tokenExpiry, [report.__dict__])
return json.dumps(embed_config.__dict__)
def get_embed_params_for_multiple_reports(self, workspace_id, report_ids, additional_dataset_ids=None):
'''Get embed params for multiple reports for a single workspace
Args:
workspace_id (str): Workspace Id
report_ids (list): Report Ids
additional_dataset_ids (list, optional): Dataset Ids which are different than the ones bound to the reports. Defaults to None.
Returns:
str: JSON representation of EmbedConfig
'''
dataset_ids = []
reports = []
for report_id in report_ids:
report_url = f'https://api.powerbi.com/v1.0/myorg/groups/{workspace_id}/reports/{report_id}'
api_response = requests.get(report_url, headers=self.get_request_header())
if api_response.status_code != 200:
raise Exception(f'Error while retrieving Embed URL\n{api_response.reason}:\t{api_response.text}\nRequestId:\t{api_response.headers.get("RequestId")}')
api_response = json.loads(api_response.text)
report_config = ReportConfig(api_response['id'], api_response['name'], api_response['embedUrl'])
reports.append(report_config.__dict__)
dataset_ids.append(api_response['datasetId'])
# Append additional dataset to the list to achieve dynamic binding later
if additional_dataset_ids is not None:
dataset_ids.extend(additional_dataset_ids)
embed_token = self.get_embed_token_for_multiple_reports_single_workspace(report_ids, dataset_ids, workspace_id)
embed_config = EmbedConfig(embed_token.tokenId, embed_token.token, embed_token.tokenExpiry, reports)
return json.dumps(embed_config.__dict__)
def get_embed_token_for_single_report_single_workspace(self, report_id, dataset_ids, target_workspace_id=None):
'''Get Embed token for single report, multiple datasets, and an optional target workspace
Args:
report_id (str): Report Id
dataset_ids (list): Dataset Ids
target_workspace_id (str, optional): Workspace Id. Defaults to None.
Returns:
EmbedToken: Embed token
'''
request_body = EmbedTokenRequestBody()
for dataset_id in dataset_ids:
request_body.datasets.append({'id': dataset_id})
request_body.reports.append({'id': report_id})
if target_workspace_id is not None:
request_body.targetWorkspaces.append({'id': target_workspace_id})
embed_token_api = 'https://api.powerbi.com/v1.0/myorg/GenerateToken'
api_response = requests.post(embed_token_api, data=json.dumps(request_body.__dict__), headers=self.get_request_header())
if api_response.status_code != 200:
raise Exception(f'Error while retrieving Embed token\n{api_response.reason}:\t{api_response.text}\nRequestId:\t{api_response.headers.get("RequestId")}')
api_response = json.loads(api_response.text)
embed_token = EmbedToken(api_response['tokenId'], api_response['token'], api_response['expiration'])
return embed_token
def get_embed_token_for_multiple_reports_single_workspace(self, report_ids, dataset_ids, target_workspace_id=None):
'''Get Embed token for multiple reports, multiple datasets, and an optional target workspace
Args:
report_ids (list): Report Ids
dataset_ids (list): Dataset Ids
target_workspace_id (str, optional): Workspace Id. Defaults to None.
Returns:
EmbedToken: Embed token
'''
request_body = EmbedTokenRequestBody()
for dataset_id in dataset_ids:
request_body.datasets.append({'id': dataset_id})
for report_id in report_ids:
request_body.reports.append({'id': report_id})
if target_workspace_id is not None:
request_body.targetWorkspaces.append({'id': target_workspace_id})
embed_token_api = 'https://api.powerbi.com/v1.0/myorg/GenerateToken'
api_response = requests.post(embed_token_api, data=json.dumps(request_body.__dict__), headers=self.get_request_header())
if api_response.status_code != 200:
raise Exception(f'Error while retrieving Embed token\n{api_response.reason}:\t{api_response.text}\nRequestId:\t{api_response.headers.get("RequestId")}')
api_response = json.loads(api_response.text)
embed_token = EmbedToken(api_response['tokenId'], api_response['token'], api_response['expiration'])
return embed_token
def get_request_header(self):
'''Get Power BI API request header
Returns:
dict: Request header
'''
return {'Content-Type': 'application/json', 'Authorization': 'Bearer ' + self.aad_service.get_access_token()}
{
"tokenId":"tokenId"
"accessToken":"accessToken==.embedToken"
"tokenExpiry":"2024-06-17T22:06:43Z"
"reportConfig":[
0:{
"reportId":"reportId"
"reportName":"Report Title"
"embedUrl":"https://app.powerbi.com/reportEmbed?reportId=reportId&groupId=groupId&w=2&config=embedToken"
"datasetId":NULL
}
]
}
Hello All,
I am using the AppOwnsData example, found here (https://github.com/microsoft/PowerBI-Developer-Samples/tree/master/Python/Embed%20for%20your%20custo...).
I have reworked the aadservice.py, pbiembedservice.py, and app.py script. I also updated the config.py script to include my own credentials.
I reworked them to use the ServicePrincipal instead of the MasterUser, and to use streamlit instead of Flask.
At first glance, it looks like it works. I get the tokenId, accessToken, tokenExpiry, reportId, reportName, and embedUrl.
However, the embedUrl does not work. It just loads and loads the Power BI Logo. I am not getting an acess or token error, or any errors for that matter so I'm not sure how to troubleshoot. When I run the original script with my config.py file, it works, so I know the credentials are all correct and that it can output the right report, however it doesn't output the report and instead the power bi logo in my modified (for streamlit) verision.
Does anyone have any clue how to fix this?
