The code retrieves the data from PBI API - it's my first one and I would like to learn how to make it pro. Here is the list of my questions:
- Can you please analyze 2 functions: GetReports and GetReportsUsers in accordance with the best practices? Should a table with metadata be added? Should a function responsible for creating the queries to API be written?
- How the mechanism of logging/troubleshooting should be implemented?
- Is the retry syntax correct?
- Any other suggestions?
And here is the code:
import azure.functions as func
import json
import requests
import pandas as pd
from azure.storage.blob import BlobServiceClient
import io
from datetime import datetime, timedelta
import logging
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient
credential = DefaultAzureCredential()
secret_client = SecretClient(vault_url='https://af-pbimonitoring.vault.azure.net', credential=credential)
api_client_secret = secret_client.get_secret('powerbi-api-admin').value
api_client_id = secret_client.get_secret('powerbi-api-admin-client-id').value
azure_tenant_id = secret_client.get_secret('azure-tenant-id').value
storage_key = secret_client.get_secret('api-storage-data').value
token_endpoint = f'https://login.microsoftonline.com/{azure_tenant_id}/oauth2/token'
resource = "https://analysis.windows.net/powerbi/api"
account_name = "afstorageapidata"
def ListGroupUsers(name, id, UserAccessRight, group_id, table, processed_groups):
# GraphAPI parameters
resource = "https://graph.microsoft.com"
# generate access token for Graph API
body_graph = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_graph)
access_token_graph = access_token_response.json().get("access_token")
headers = {
"Authorization": f"Bearer {access_token_graph}"
}
response = requests.get(f"https://graph.microsoft.com/v1.0/groups/{group_id}/members", headers=headers)
users_data = response.json()
# add each group member to list
for user in users_data['value']:
user_type = 'Member'
principal_type = 'User'
if user["@odata.type"] == "#microsoft.graph.group":
user_type = None
principal_type = 'Group'
else:
principal_type = 'None'
table.append({
'id': id,
'name': name,
'identifier': user['id'],
'UserAccessRight': UserAccessRight,
'displayName': user['displayName'],
'principalType': principal_type,
'emailAddress': user['mail'],
'graphId': user['id'],
'userType': user_type,
'requestDate' : datetime.now().strftime('%Y-%m-%d %H:%M:%S')
})
# if member is a group and it haven't been listed already use recursion to list group members
if user.get('@odata.type') == "#microsoft.graph.group" and user['id'] not in processed_groups:
processed_groups.append(user['id'])
return (ListGroupUsers(name, id, UserAccessRight, user['id'], table, processed_groups))
return table
app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)
@app.route(route="GetReportUsers")
def GetReportUsers(req: func.HttpRequest) -> func.HttpResponse:
logging.info('HTTP trigger function started. Invoking Power BI API..,')
body = req.get_body()
json_data = json.loads(body.decode('utf-8'))
# generate access token for Power BI API
body_admin = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_admin)
access_token_admin = access_token_response.json().get("access_token")
data = []
# for each report
for item in json_data["value"]:
try:
report_name = item["reportName"]
report_id = item["reportId"]
# list report users
api_url = f"https://api.powerbi.com/v1.0/myorg/admin/reports/{report_id}/users"
headers = {
"Authorization": f"Bearer {access_token_admin}"
}
response = requests.get(api_url, headers=headers)
if response.status_code == 200:
users_data = response.json()
for user in users_data['value']:
data.append({
'id': report_id,
'name': report_name,
'identifier': user['identifier'],
'UserAccessRight': user['reportUserAccessRight'],
'displayName': user['displayName'],
'principalType': user['principalType'],
'emailAddress': user['identifier'],
'graphId': user['graphId'],
'userType': user.get('userType')
})
# use function to list group members
if user["principalType"] == "Group":
data += ListGroupUsers(report_name, report_id, user['reportUserAccessRight'], user["identifier"], [], [user['graphId']])
except:
logging.info(f"Report {item} didn't process successfully")
logging.info("Data generated successfully. Starting connection to blob storage...")
# blob storage parameters
today = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
df = pd.DataFrame(data)
df['requestDate'] = today
today = datetime.now().strftime('%Y-%m-%d')
container_name = "staging"
blob_path = f"reportusers/{today}/reportusers.csv"
# connect to blob
blob_service_client = BlobServiceClient(account_url=f"https://{account_name}.blob.core.windows.net", credential=storage_key)
container_client = blob_service_client.get_container_client(container_name)
blob_client = container_client.get_blob_client(blob_path)
# save dataframe to csv
csv_data = df.to_csv(index=False)
csv_bytes = csv_data.encode("utf-8")
# upload file to blob
with io.BytesIO(csv_bytes) as upload:
blob_client.upload_blob(upload, overwrite=True)
return func.HttpResponse(f"{data}")
@app.route(route="GetBearerToken")
def GetBearerToken(req: func.HttpRequest) -> func.HttpResponse:
# generate access token for Power BI API
body_admin = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_admin)
access_token_admin = access_token_response.json().get("access_token")
return func.HttpResponse(f'{{ "access_token": "{access_token_admin}" }}', mimetype="application/json")
@app.route(route="GetReports", auth_level=func.AuthLevel.FUNCTION)
def GetReports(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Python HTTP trigger function processed a request.')
body_admin = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_admin)
access_token_admin = access_token_response.json().get("access_token")
api_url = f"https://api.powerbi.com/v1.0/myorg/admin/reports"
headers = {
"Authorization": f"Bearer {access_token_admin}"
}
response = requests.get(api_url, headers=headers)
reports_json = response.json()
data = []
for report in reports_json['value']:
data.append({
'id': report.get('id'),
'reportType': report.get('reportType'),
'name': report.get('name'),
'webUrl': report.get('webUrl'),
'embedUrl': report.get('embedUrl'),
'datasetId': report.get('datasetId'),
'createdDateTime': report.get('createdDateTime'),
'modifiedDateTime': report.get('modifiedDateTime'),
'modifiedBy': report.get('modifiedBy'),
'createdBy': report.get('createdBy'),
'users': report.get('users'),
'subscriptions': report.get('subscriptions'),
'workspaceId': report.get('workspaceId')
})
logging.info("Data generated successfully. Starting connection to blob storage...")
# blob storage parameters
today = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
df = pd.DataFrame(data)
df['requestDate'] = today
today = datetime.now().strftime('%Y-%m-%d')
container_name = "staging"
blob_path = f"reports/{today}/reports.csv"
# connect to blob
blob_service_client = BlobServiceClient(account_url=f"https://{account_name}.blob.core.windows.net", credential=storage_key)
container_client = blob_service_client.get_container_client(container_name)
blob_client = container_client.get_blob_client(blob_path)
# save dataframe to csv
csv_data = df.to_csv(index=False)
csv_bytes = csv_data.encode("utf-8")
# upload file to blob
with io.BytesIO(csv_bytes) as upload:
blob_client.upload_blob(upload, overwrite=True)
return func.HttpResponse(f"{data}")
@app.route(route="GetApps", auth_level=func.AuthLevel.FUNCTION)
def GetApps(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Python HTTP trigger function processed a request.')
body_admin = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_admin)
access_token_admin = access_token_response.json().get("access_token")
api_url = f"https://api.powerbi.com/v1.0/myorg/admin/apps?$top=1000"
headers = {
"Authorization": f"Bearer {access_token_admin}"
}
response = requests.get(api_url, headers=headers)
apps_json = response.json()
data = []
for app in apps_json['value']:
data.append({
'id': app.get('id'),
'name': app.get('name'),
'lastUpdate': app.get('lastUpdate'),
'description': app.get('description'),
'publishedBy': app.get('publishedBy'),
'workspaceId': app.get('workspaceId'),
'users': app.get('users')
})
logging.info("Data generated successfully. Starting connection to blob storage...")
# blob storage parameters
today = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
df = pd.DataFrame(data)
df['requestDate'] = today
today = datetime.now().strftime('%Y-%m-%d')
container_name = "staging"
blob_path = f"apps/{today}/apps.csv"
# connect to blob
blob_service_client = BlobServiceClient(account_url=f"https://{account_name}.blob.core.windows.net", credential=storage_key)
container_client = blob_service_client.get_container_client(container_name)
blob_client = container_client.get_blob_client(blob_path)
# save dataframe to csv
csv_data = df.to_csv(index=False)
csv_bytes = csv_data.encode("utf-8")
# upload file to blob
with io.BytesIO(csv_bytes) as upload:
blob_client.upload_blob(upload, overwrite=True)
return func.HttpResponse(f"{data}")
@app.route(route="GetGroups", auth_level=func.AuthLevel.FUNCTION)
def GetGroups(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Python HTTP trigger function processed a request.')
body_admin = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_admin)
access_token_admin = access_token_response.json().get("access_token")
api_url = f"https://api.powerbi.com/v1.0/myorg/admin/groups?$top=5000"
headers = {
"Authorization": f"Bearer {access_token_admin}"
}
response = requests.get(api_url, headers=headers)
groups_json = response.json()
data = []
for group in groups_json['value']:
data.append({
'id': group.get('id'),
'isReadOnly': group.get('isReadOnly'),
'isOnDedicatedCapacity': group.get('isOnDedicatedCapacity'),
'capacityMigrationStatus': group.get('capacityMigrationStatus'),
'type': group.get('type'),
'state': group.get('state'),
'hasWorkspaceLevelSettings': group.get('hasWorkspaceLevelSettings'),
'name': group.get('name')
})
logging.info("Data generated successfully. Starting connection to blob storage...")
# blob storage parameters
today = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
df = pd.DataFrame(data)
df['requestDate'] = today
today = datetime.now().strftime('%Y-%m-%d')
container_name = "staging"
blob_path = f"groups/{today}/groups.csv"
# connect to blob
blob_service_client = BlobServiceClient(account_url=f"https://{account_name}.blob.core.windows.net", credential=storage_key)
container_client = blob_service_client.get_container_client(container_name)
blob_client = container_client.get_blob_client(blob_path)
# save dataframe to csv
csv_data = df.to_csv(index=False)
csv_bytes = csv_data.encode("utf-8")
# upload file to blob
with io.BytesIO(csv_bytes) as upload:
blob_client.upload_blob(upload, overwrite=True)
return func.HttpResponse(f"{data}")
@app.route(route="GetDatasets", auth_level=func.AuthLevel.FUNCTION)
def GetDatasets(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Python HTTP trigger function processed a request.')
body_admin = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_admin)
access_token_admin = access_token_response.json().get("access_token")
api_url = f"https://api.powerbi.com/v1.0/myorg/admin/datasets"
headers = {
"Authorization": f"Bearer {access_token_admin}"
}
response = requests.get(api_url, headers=headers)
datasets_json = response.json()
data = []
for dataset in datasets_json['value']:
data.append({
'id': dataset.get('id'),
'name': dataset.get('name'),
'webUrl': dataset.get('webUrl'),
'addRowsAPIEnabled': dataset.get('addRowsAPIEnabled'),
'configuredBy': dataset.get('configuredBy'),
'isRefreshable': dataset.get('isRefreshable'),
'isEffectiveIdentityRequired': dataset.get('isEffectiveIdentityRequired'),
'isEffectiveIdentityRolesRequired': dataset.get('isEffectiveIdentityRolesRequired'),
'targetStorageMode': dataset.get('targetStorageMode'),
'createdDate': dataset.get('createdDate'),
'contentProviderType': dataset.get('contentProviderType'),
'createReportEmbedURL': dataset.get('createReportEmbedURL'),
'qnaEmbedURL': dataset.get('qnaEmbedURL'),
'upstreamDatasets': dataset.get('upstreamDatasets'),
'users': dataset.get('users'),
'isInPlaceSharingEnabled': dataset.get('isInPlaceSharingEnabled'),
'workspaceId': dataset.get('workspaceId'),
'autoSyncReadOnlyReplicas': dataset.get('queryScaleOutSettings').get('autoSyncReadOnlyReplicas'),
'maxReadOnlyReplicas': dataset.get('queryScaleOutSettings').get('maxReadOnlyReplicas')
})
logging.info("Data generated successfully. Starting connection to blob storage...")
# blob storage parameters
today = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
df = pd.DataFrame(data)
df['requestDate'] = today
today = datetime.now().strftime('%Y-%m-%d')
container_name = "staging"
blob_path = f"datasets/{today}/datasets.csv"
# connect to blob
blob_service_client = BlobServiceClient(account_url=f"https://{account_name}.blob.core.windows.net", credential=storage_key)
container_client = blob_service_client.get_container_client(container_name)
blob_client = container_client.get_blob_client(blob_path)
# save dataframe to csv
csv_data = df.to_csv(index=False)
csv_bytes = csv_data.encode("utf-8")
# upload file to blob
with io.BytesIO(csv_bytes) as upload:
blob_client.upload_blob(upload, overwrite=True)
return func.HttpResponse(f"{data}")
@app.route(route="GetAppUsers", auth_level=func.AuthLevel.FUNCTION)
def GetAppUsers(req: func.HttpRequest) -> func.HttpResponse:
logging.info('HTTP trigger function started. Invoking Power BI API..,')
body = req.get_body()
json_data = json.loads(body.decode('utf-8'))
# generate access token for Power BI API
body_admin = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_admin)
access_token_admin = access_token_response.json().get("access_token")
data = []
# for each report
for item in json_data["value"]:
try:
app_name = item["name"]
app_id = item["id"]
# list report users
api_url = f"https://api.powerbi.com/v1.0/myorg/admin/apps/{app_id}/users"
headers = {
"Authorization": f"Bearer {access_token_admin}"
}
response = requests.get(api_url, headers=headers)
if response.status_code == 200:
users_data = response.json()
for user in users_data['value']:
data.append({
'id': app_id,
'name': app_name,
'UserAccessRight': user.get('appUserAccessRight'),
'emailAddress': user.get('emailAddress'),
'displayName': user.get('displayName'),
'identifier': user.get('identifier'),
'graphId': user.get('graphId'),
'principalType': user.get('principalType'),
'userType': user.get('userType')
})
# use function to list group members
if user["principalType"] == "Group":
data += ListGroupUsers(app_name, app_id, user['appUserAccessRight'], user["identifier"], [], [user['graphId']])
except:
logging.info(f"App {item} didn't process successfully")
logging.info("Data generated successfully. Starting connection to blob storage...")
# blob storage parameters
today = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
df = pd.DataFrame(data)
df['requestDate'] = today
today = datetime.now().strftime('%Y-%m-%d')
container_name = "staging"
blob_path = f"appusers/{today}/appusers.csv"
# connect to blob
blob_service_client = BlobServiceClient(account_url=f"https://{account_name}.blob.core.windows.net", credential=storage_key)
container_client = blob_service_client.get_container_client(container_name)
blob_client = container_client.get_blob_client(blob_path)
# save dataframe to csv
csv_data = df.to_csv(index=False)
csv_bytes = csv_data.encode("utf-8")
# upload file to blob
with io.BytesIO(csv_bytes) as upload:
blob_client.upload_blob(upload, overwrite=True)
return func.HttpResponse(f"{data}")
@app.route(route="GetGroupsUsers", auth_level=func.AuthLevel.FUNCTION)
def GetGroupsUsers(req: func.HttpRequest) -> func.HttpResponse:
logging.info('HTTP trigger function started. Invoking Power BI API..,')
body = req.get_body()
json_data = json.loads(body.decode('utf-8'))
# generate access token for Power BI API
body_admin = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_admin)
access_token_admin = access_token_response.json().get("access_token")
data = []
# for each report
for item in json_data["value"]:
logging.info(f'{item}')
try:
group_name = item["name"]
group_id = item["id"]
# list report users
api_url = f"https://api.powerbi.com/v1.0/myorg/admin/groups/{group_id}/users"
headers = {
"Authorization": f"Bearer {access_token_admin}"
}
response = requests.get(api_url, headers=headers)
if response.status_code == 200:
users_data = response.json()
for user in users_data['value']:
data.append({
'id': group_id,
'name': group_name,
'UserAccessRight': user.get('groupUserAccessRight'),
'emailAddress': user.get('emailAddress'),
'displayName': user.get('displayName'),
'identifier': user.get('identifier'),
'graphId': user.get('graphId'),
'principalType': user.get('principalType'),
'userType': user.get('userType')
})
# use function to list group members
if user["principalType"] == "Group":
data += ListGroupUsers(group_name, group_id, user['groupUserAccessRight'], user["identifier"], [], [user['graphId']])
except:
logging.info(f"Group {item} didn't process successfully")
logging.info("Data generated successfully. Starting connection to blob storage...")
# blob storage parameters
today = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
df = pd.DataFrame(data)
df['requestDate'] = today
today = datetime.now().strftime('%Y-%m-%d')
container_name = "staging"
blob_path = f"groupusers/{today}/groupusers.csv"
# connect to blob
blob_service_client = BlobServiceClient(account_url=f"https://{account_name}.blob.core.windows.net", credential=storage_key)
container_client = blob_service_client.get_container_client(container_name)
blob_client = container_client.get_blob_client(blob_path)
# save dataframe to csv
csv_data = df.to_csv(index=False)
csv_bytes = csv_data.encode("utf-8")
# upload file to blob
with io.BytesIO(csv_bytes) as upload:
blob_client.upload_blob(upload, overwrite=True)
return func.HttpResponse(f"{data}")
@app.route(route="GetGraphGroups", auth_level=func.AuthLevel.FUNCTION)
def GetGraphGroups(req: func.HttpRequest) -> func.HttpResponse:
logging.info('HTTP trigger function started. Invoking Power BI API..,')
# Graph API parameters
resource = "https://graph.microsoft.com"
# generate access token for Graph API
body_graph = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_graph)
access_token_graph = access_token_response.json().get("access_token")
headers = {
"Authorization": f"Bearer {access_token_graph}"
}
response = requests.get(f"https://graph.microsoft.com/v1.0/groups/?$select=id,displayName", headers=headers)
groups_json = response.json()
data = []
for group in groups_json['value']:
data.append({
'id': group.get('id'),
'displayName': group.get('displayName')
})
logging.info("Data generated successfully. Starting connection to blob storage...")
# blob storage parameters
today = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
df = pd.DataFrame(data)
df['requestDate'] = today
today = datetime.now().strftime('%Y-%m-%d')
container_name = "staging"
blob_path = f"graphgroups/{today}/graphgroups.csv"
# connect to blob
blob_service_client = BlobServiceClient(account_url=f"https://{account_name}.blob.core.windows.net", credential=storage_key)
container_client = blob_service_client.get_container_client(container_name)
blob_client = container_client.get_blob_client(blob_path)
# save dataframe to csv
csv_data = df.to_csv(index=False)
csv_bytes = csv_data.encode("utf-8")
# upload file to blob
with io.BytesIO(csv_bytes) as upload:
blob_client.upload_blob(upload, overwrite=True)
return func.HttpResponse(f"{data}")
@app.route(route="GetGraphUsers", auth_level=func.AuthLevel.FUNCTION)
def GetGraphUsers(req: func.HttpRequest) -> func.HttpResponse:
logging.info('HTTP trigger function started. Invoking Power BI API..,')
resource = "https://graph.microsoft.com"
# generate access token for Graph API
body_graph = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_graph)
access_token_graph = access_token_response.json().get("access_token")
headers = {
"Authorization": f"Bearer {access_token_graph}"
}
response = requests.get(f"https://graph.microsoft.com/v1.0/users?$select=id,displayName,givenName,surname,mail,mailNickname,jobTitle,department,companyName,officeLocation,city,accountEnabled", headers=headers)
users_json = response.json()
data = []
for user in users_json['value']:
data.append({
'id': user.get('id'),
'displayName': user.get('displayName'),
'givenName': user.get('givenName'),
'surname': user.get('surname'),
'mail': user.get('mail'),
'mailNickname': user.get('mailNickname'),
'jobTitle': user.get('jobTitle'),
'department': user.get('department'),
'companyName': user.get('companyName'),
'officeLocation': user.get('officeLocation'),
'city': user.get('city'),
'accountEnabled': user.get('accountEnabled')
})
logging.info("Data generated successfully. Starting connection to blob storage...")
# blob storage parameters
today = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
df = pd.DataFrame(data)
df['requestDate'] = today
today = datetime.now().strftime('%Y-%m-%d')
container_name = "staging"
blob_path = f"graphusers/{today}/graphusers.csv"
# connect to blob
blob_service_client = BlobServiceClient(account_url=f"https://{account_name}.blob.core.windows.net", credential=storage_key)
container_client = blob_service_client.get_container_client(container_name)
blob_client = container_client.get_blob_client(blob_path)
# save dataframe to csv
csv_data = df.to_csv(index=False)
csv_bytes = csv_data.encode("utf-8")
# upload file to blob
with io.BytesIO(csv_bytes) as upload:
blob_client.upload_blob(upload, overwrite=True)
return func.HttpResponse(f"{data}")
@app.route(route="GetGraphUsersAssignedPlans", auth_level=func.AuthLevel.FUNCTION)
def GetGraphUsersAssignedPlans(req: func.HttpRequest) -> func.HttpResponse:
logging.info('HTTP trigger function started. Invoking Power BI API..,')
resource = "https://graph.microsoft.com"
# generate access token for Graph API
body_graph = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_graph)
access_token_graph = access_token_response.json().get("access_token")
headers = {
"Authorization": f"Bearer {access_token_graph}"
}
response = requests.get(f"https://graph.microsoft.com/v1.0/users?$select=id,assignedPlans", headers=headers)
users_json = response.json()
data = []
for user in users_json['value']:
data.append({
'id': user.get('id'),
'items' : user.get('assignedPlans')
})
logging.info("Data generated successfully. Starting connection to blob storage...")
# blob storage parameters
today = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
df = pd.DataFrame(data)
df['requestDate'] = today
today = datetime.now().strftime('%Y-%m-%d')
container_name = "staging"
blob_path = f"graphusersassignedplans/{today}/graphusersassignedplans.csv"
# connect to blob
blob_service_client = BlobServiceClient(account_url=f"https://{account_name}.blob.core.windows.net", credential=storage_key)
container_client = blob_service_client.get_container_client(container_name)
blob_client = container_client.get_blob_client(blob_path)
# save dataframe to csv
csv_data = df.to_csv(index=False)
csv_bytes = csv_data.encode("utf-8")
# upload file to blob
with io.BytesIO(csv_bytes) as upload:
blob_client.upload_blob(upload, overwrite=True)
return func.HttpResponse(f"{data}")
@app.route(route="GetGroupReports", auth_level=func.AuthLevel.FUNCTION)
def GetGroupReports(req: func.HttpRequest) -> func.HttpResponse:
logging.info('HTTP trigger function started. Invoking Power BI API..,')
body = req.get_body()
json_data = json.loads(body.decode('utf-8'))
# generate access token for Power BI API
body_admin = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_admin)
access_token_admin = access_token_response.json().get("access_token")
data = []
# for each report
for item in json_data["value"]:
group_id = item["workspaceId"]
group_name = item["workspaceName"]
# list report users
api_url = f'https://api.powerbi.com/v1.0/myorg/admin/groups/{group_id}/reports'
headers = {
"Authorization": f"Bearer {access_token_admin}"
}
response = requests.get(api_url, headers=headers)
groups_json = response.json()
for report in groups_json['value']:
data.append({
'workspaceId' : group_id,
'workspaceName' : group_name,
'id': report.get('id'),
'reportType': report.get('reportType'),
'name': report.get('name'),
'webUrl': report.get('webUrl'),
'embedUrl': report.get('embedUrl'),
'datasetId': report.get('datasetId'),
'createdDateTime': report.get('createdDateTime'),
'modifiedDateTime': report.get('modifiedDateTime'),
'modifiedBy': report.get('modifiedBy'),
'createdBy': report.get('createdBy'),
'endorsement': report.get('endorsementDetails'),
'users': report.get('users'),
'subscriptions': report.get('subscriptions')
})
logging.info("Data generated successfully. Starting connection to blob storage...")
# blob storage parameters
today = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
df = pd.DataFrame(data)
df['requestDate'] = today
today = datetime.now().strftime('%Y-%m-%d')
container_name = "staging"
blob_path = f"groupreports/{today}/groupreports.csv"
# connect to blob
blob_service_client = BlobServiceClient(account_url=f"https://{account_name}.blob.core.windows.net", credential=storage_key)
container_client = blob_service_client.get_container_client(container_name)
blob_client = container_client.get_blob_client(blob_path)
# save dataframe to csv
csv_data = df.to_csv(index=False)
csv_bytes = csv_data.encode("utf-8")
# upload file to blob
with io.BytesIO(csv_bytes) as upload:
blob_client.upload_blob(upload, overwrite=True)
return func.HttpResponse(f"{data}")
@app.route(route="GetAppReports", auth_level=func.AuthLevel.FUNCTION)
def GetAppReports(req: func.HttpRequest) -> func.HttpResponse:
logging.info('HTTP trigger function started. Invoking Power BI API..,')
body = req.get_body()
json_data = json.loads(body.decode('utf-8'))
# generate access token for Power BI API
body_admin = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_admin)
access_token_admin = access_token_response.json().get("access_token")
data = []
# for each report
for item in json_data["value"]:
app_id = item["appId"]
app_name = item["appName"]
# list report users
api_url = f'https://api.powerbi.com/v1.0/myorg/admin/apps/{app_id}/reports'
headers = {
"Authorization": f"Bearer {access_token_admin}"
}
response = requests.get(api_url, headers=headers)
apps_json = response.json()
for report in apps_json['value']:
data.append({
'appId' : app_id,
'appName' : app_name,
'id': report.get('id'),
'reportType': report.get('reportType'),
'name': report.get('name'),
'webUrl': report.get('webUrl'),
'embedUrl': report.get('embedUrl'),
'datasetId': report.get('datasetId'),
'createdDateTime': report.get('createdDateTime'),
'modifiedDateTime': report.get('modifiedDateTime'),
'modifiedBy': report.get('modifiedBy'),
'createdBy': report.get('createdBy'),
'users': report.get('users'),
'subscriptions': report.get('subscriptions'),
'workspaceId': report.get('workspaceId')
})
logging.info("Data generated successfully. Starting connection to blob storage...")
# blob storage parameters
today = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
df = pd.DataFrame(data)
df['requestDate'] = today
today = datetime.now().strftime('%Y-%m-%d')
container_name = "staging"
blob_path = f"appreports/{today}/appreports.csv"
# connect to blob
blob_service_client = BlobServiceClient(account_url=f"https://{account_name}.blob.core.windows.net", credential=storage_key)
container_client = blob_service_client.get_container_client(container_name)
blob_client = container_client.get_blob_client(blob_path)
# save dataframe to csv
csv_data = df.to_csv(index=False)
csv_bytes = csv_data.encode("utf-8")
# upload file to blob
with io.BytesIO(csv_bytes) as upload:
blob_client.upload_blob(upload, overwrite=True)
return func.HttpResponse(f"{data}")
@app.route(route="GetActivityEvents", auth_level=func.AuthLevel.FUNCTION)
def GetActivityEvents(req: func.HttpRequest) -> func.HttpResponse:
logging.info('HTTP trigger function started. Invoking Power BI API..,')
body = req.get_body()
json_data = json.loads(body.decode('utf-8'))
# generate access token for Power BI API
body_admin = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_admin)
access_token_admin = access_token_response.json().get("access_token")
data = []
yesterday_start = (datetime.now() - timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
yesterday_end = (datetime.now() - timedelta(days=1)).replace(hour=23, minute=59, second=59, microsecond=0)
# for each report
for item in json_data["value"]:
activity_type = item["ActivityType"]
# list report users
api_url = f"https://api.powerbi.com/v1.0/myorg/admin/activityevents?startDateTime='{yesterday_start.isoformat()}Z'&endDateTime='{yesterday_end.isoformat()}Z'&$filter=Activity eq '{activity_type}'"
headers = {
"Authorization": f"Bearer {access_token_admin}"
}
response = requests.get(api_url, headers=headers)
activities_json = response.json()
for activity in activities_json['activityEventEntities']:
data.append({
"Id": activity["Id"],
"RecordType": activity["RecordType"],
"CreationTime": activity["CreationTime"],
"Operation": activity["Operation"],
"OrganizationId": activity["OrganizationId"],
"UserType": activity["UserType"],
"UserKey": activity["UserKey"],
"Workload": activity["Workload"],
"UserId": activity["UserId"],
"ClientIP": activity["ClientIP"],
"Activity": activity["Activity"],
"ItemName": activity.get("ItemName"),
"WorkSpaceName": activity.get("WorkSpaceName"),
"DatasetName": activity.get("DatasetName"),
"ReportName": activity.get("ReportName"),
"CapacityId": activity.get("CapacityId"),
"CapacityName": activity.get("CapacityName"),
"WorkspaceId": activity.get("WorkspaceId"),
"ObjectId": activity.get("ObjectId"),
"DatasetId": activity.get("DatasetId"),
"ReportId": activity.get("ReportId"),
"ArtifactId": activity.get("ArtifactId"),
"ArtifactName": activity.get("ArtifactName"),
"IsSuccess": activity["IsSuccess"],
"ReportType": activity.get("ReportType"),
"RequestId": activity["RequestId"],
"ActivityId": activity["ActivityId"],
"DistributionMethod": activity.get("DistributionMethod"),
"ConsumptionMethod": activity.get("ConsumptionMethod"),
"ArtifactKind": activity.get("ArtifactKind"),
"RefreshEnforcementPolicy": activity.get("RefreshEnforcementPolicy")
})
logging.info("Data generated successfully. Starting connection to blob storage...")
# blob storage parameters
today = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
df = pd.DataFrame(data)
df['requestDate'] = today
today = datetime.now().strftime('%Y-%m-%d')
container_name = "staging"
blob_path = f"activityevents/{today}/activityevents.csv"
# connect to blob
blob_service_client = BlobServiceClient(account_url=f"https://{account_name}.blob.core.windows.net", credential=storage_key)
container_client = blob_service_client.get_container_client(container_name)
blob_client = container_client.get_blob_client(blob_path)
# save dataframe to csv
csv_data = df.to_csv(index=False)
csv_bytes = csv_data.encode("utf-8")
# upload file to blob
with io.BytesIO(csv_bytes) as upload:
blob_client.upload_blob(upload, overwrite=True)
return func.HttpResponse(f"{data}")
@app.route(route="GetDatasources", auth_level=func.AuthLevel.FUNCTION)
def GetDatasources(req: func.HttpRequest) -> func.HttpResponse:
logging.info('HTTP trigger function started. Invoking Power BI API..,')
body = req.get_body()
json_data = json.loads(body.decode('utf-8'))
# generate access token for Power BI API
body_admin = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_admin)
access_token_admin = access_token_response.json().get("access_token")
data = []
# for each report
for item in json_data["value"]:
dataset_id = item["datasetId"]
# list report users
api_url = f'https://api.powerbi.com/v1.0/myorg/admin/datasets/{dataset_id}/datasources'
headers = {
"Authorization": f"Bearer {access_token_admin}"
}
response = requests.get(api_url, headers=headers)
datasets_json = response.json()
for datasource in datasets_json['value']:
data.append({
'datasetId' : dataset_id,
'datasourceType': datasource.get('datasourceType'),
'server': datasource['connectionDetails'].get('server'),
'database': datasource['connectionDetails'].get('database'),
'datasourceId': datasource.get('datasourceId'),
'gatewayId': datasource.get('gatewayId')
})
logging.info("Data generated successfully. Starting connection to blob storage...")
# blob storage parameters
today = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
df = pd.DataFrame(data)
df['requestDate'] = today
today = datetime.now().strftime('%Y-%m-%d')
container_name = "staging"
blob_path = f"datasources/{today}/datasources.csv"
# connect to blob
blob_service_client = BlobServiceClient(account_url=f"https://{account_name}.blob.core.windows.net", credential=storage_key)
container_client = blob_service_client.get_container_client(container_name)
blob_client = container_client.get_blob_client(blob_path)
# save dataframe to csv
csv_data = df.to_csv(index=False)
csv_bytes = csv_data.encode("utf-8")
# upload file to blob
with io.BytesIO(csv_bytes) as upload:
blob_client.upload_blob(upload, overwrite=True)
return func.HttpResponse(f"{data}")
@app.route(route="GetDatasetRefreshes", auth_level=func.AuthLevel.FUNCTION)
def GetDatasetRefreshes(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Python HTTP trigger function processed a request.')
body_admin = {
"grant_type": "client_credentials",
"client_id": api_client_id,
"client_secret": api_client_secret,
"resource": resource
}
access_token_response = requests.post(token_endpoint, data=body_admin)
access_token_admin = access_token_response.json().get("access_token")
api_url = f"https://api.powerbi.com/v1.0/myorg/admin/capacities/refreshables?$top=5000"
headers = {
"Authorization": f"Bearer {access_token_admin}"
}
response = requests.get(api_url, headers=headers)
datasets_json = response.json()
data = []
for refreshable in datasets_json['value']:
data.append({
'id': refreshable.get('id'),
'name': refreshable.get('name'),
'kind': refreshable.get('kind'),
'id': refreshable['lastRefresh'].get('id'),
'refreshType': refreshable['lastRefresh'].get('refreshType'),
'startTime': refreshable['lastRefresh'].get('startTime'),
'endTime': refreshable['lastRefresh'].get('endTime'),
'serviceExceptionJson': refreshable['lastRefresh'].get('serviceExceptionJson'),
'status': refreshable['lastRefresh'].get('status'),
'requestId': refreshable['lastRefresh'].get('requestId'),
'extendedStatus': refreshable['lastRefresh'].get('extendedStatus'),
'refreshAttempts': refreshable['lastRefresh'].get('refreshAttempts'),
'days': refreshable['refreshSchedule'].get('days'),
'times': refreshable['refreshSchedule'].get('times'),
'enabled': refreshable['refreshSchedule'].get('enabled'),
'localTimeZoneId': refreshable['refreshSchedule'].get('localTimeZoneId'),
'notifyOption': refreshable['refreshSchedule'].get('notifyOption'),
'configuredBy': refreshable.get('configuredBy')
})
logging.info("Data generated successfully. Starting connection to blob storage...")
# blob storage parameters
today = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
df = pd.DataFrame(data)
df['requestDate'] = today
today = datetime.now().strftime('%Y-%m-%d')
container_name = "staging"
blob_path = f"datasetrefreshes/{today}/datasetrefreshes.csv"
# connect to blob
blob_service_client = BlobServiceClient(account_url=f"https://{account_name}.blob.core.windows.net", credential=storage_key)
container_client = blob_service_client.get_container_client(container_name)
blob_client = container_client.get_blob_client(blob_path)
# save dataframe to csv
csv_data = df.to_csv(index=False)
csv_bytes = csv_data.encode("utf-8")
# upload file to blob
with io.BytesIO(csv_bytes) as upload:
blob_client.upload_blob(upload, overwrite=True)
return func.HttpResponse(f"{data}")
```
