Find everything you need to get certified on Fabric—skills challenges, live sessions, exam prep, role guidance, and more. Get started
Here's the Python code for Workspace Check and Creation, Assign to Capacity and Lakehouse Creation and User access:
#############################################
# Import libraries and functions
import json
import requests
import datetime, time
from datetime import *
from time import *
from pyspark.sql.types import *
from pyspark.sql.functions import *
# Parameters
dest_workspacename = 'fabric_ws'
dest_lakehousename = 'fabric_lh'
capacityname = 'fabric capacity' #fabric Capacity name
fabric_base_url = 'https://api.fabric.microsoft.com/v1/'
powerbi_base_url = 'https://api.powerbi.com/v1.0/myorg/groups/'
keyvault = 'https://kvl.vault.azure.net/' # Keyvault Name
# Getting information for Tenant ID, Client ID (Service Principal) and Client Password
tenant_id=mssparkutils.credentials.getSecret(keyvault, 'tenant_id')
client_id=mssparkutils.credentials.getSecret(keyvault, 'client_id')
client_secret=mssparkutils.credentials.getSecret(keyvault, 'client_secret')
# Rest API response codes
http_ok_code = 200
http_accepted_code = 202
http_created_code = 201
access_token = None
token_expiration = None
admin_groups=[]
# Token Generation
def token_generation(client_id,client_secret,tenant_id):
token_endpoint = f'https://login.microsoftonline.com/{tenant_id}/oauth2/token'
resource_endpoint = 'https://analysis.windows.net/powerbi/api'
token_payload = {
'grant_type': 'client_credentials',
'client_id': client_id,
'client_secret': client_secret,
'resource': resource_endpoint
}
response = requests.post(token_endpoint, data=token_payload)
if response.status_code == 200:
token_data = json.loads(response.text)
token = token_data.get('access_token')
access_token = f'Bearer {token}'
expires_on = token_data.get('expires_on')
if expires_on is not None:
token_expiration = datetime.utcfromtimestamp(int(expires_on)) + timedelta(hours=1)
else:
token_expiration = None
else:
access_token = None
token_expiration = None
raise requests.HTTPError(response.text)
return access_token,token_expiration
if access_token is None or token_expiration is None or token_expiration < datetime.utcnow():
access_token, token_expiration = token_generation(client_id, client_secret, tenant_id)
# Workspace check and creation
url = fabric_base_url+"workspaces"
payload={}
headers = {'Authorization': access_token}
response = requests.request("GET", url, headers=headers, data=payload)
found = False
if response.status_code == http_ok_code:
response=json.loads(response.text)["value"]
for row in response:
if row['displayName'] == dest_workspacename:
found = True
dest_workspaceid=row['id']
if not found:
payload = f'{{"displayName": "{dest_workspacename}"}}'
headers = {'Authorization': access_token,'Content-Type': 'application/json'}
response = requests.request("POST", url, headers=headers, data=payload)
if response.status_code == http_ok_code or response.status_code == http_accepted_code or response.status_code == http_created_code:
dest_workspaceid = json.loads(response.text)["id"]
else:
dest_workspaceid=None
raise requests.HTTPError(response.text)
# Getting capacity ID and Assign Capacity to Workspace
url_capacities = fabric_base_url+"capacities"
payload={}
headers = {'Authorization': access_token}
response = requests.request("GET", url_capacities, headers=headers, data=payload)
if response.status_code == http_ok_code:
response=json.loads(response.text)["value"]
for row in response:
if row['displayName'] == capacityname:
found = True
capacityid=row['id']
if not found:
capacityid=None
raise requests.HTTPError(response)
url_capacity = fabric_base_url+f"workspaces/{dest_workspaceid}/assignToCapacity"
payload = f'{{"capacityId": "{capacityid}"}}'
headers = {'Authorization': access_token,'Content-Type': 'application/json'}
response = requests.request("POST", url_capacity, headers=headers, data=payload)
if response.status_code == http_ok_code or response.status_code == http_accepted_code or response.status_code == http_created_code:
print("Success!")
else:
print("failed")
raise requests.HTTPError(response)
# Lakehouse check and creation
url_lakehouse = fabric_base_url+f"workspaces/{dest_workspaceid}/lakehouses"
payload={}
headers = {'Authorization': access_token}
response = requests.request("GET", url_lakehouse, headers=headers, data=payload)
found = False
if response.status_code == http_ok_code:
response=json.loads(response.text)["value"]
for row in response:
if row['displayName'] == dest_lakehousename:
found = True
dest_lakehouseid=row['id']
if not found:
payload = f'{{"displayName": "{dest_lakehousename}"}}'
headers = {'Authorization': access_token,'Content-Type': 'application/json'}
response = requests.request("POST", url_lakehouse, headers=headers, data=payload)
if response.status_code == http_ok_code or response.status_code == http_accepted_code or response.status_code == http_created_code:
dest_lakehouseid = json.loads(response.text)["id"]
else:
dest_lakehouseid = None
raise requests.HTTPError(response.text)
# Getting SQL Endpoint details
dest_sqlendpointlink = None
while dest_sqlendpointlink == None or dest_sqlendpointlink == "":
sleep(30)
url_lakehouse_details = url_lakehouse + f"/{dest_lakehouseid}"
response = requests.request("GET", url_lakehouse_details, headers=headers, data=payload)
if response.status_code == http_ok_code or response.status_code == http_accepted_code or response.status_code == http_created_code:
dest_sqlendpointlink = json.loads(response.text)['properties']['sqlEndpointProperties']['connectionString']
else:
dest_sqlendpointlink = None
raise requests.HTTPError(response.text)
# Getting User details:
url_users = powerbi_base_url+f'{dest_workspaceid}/users'
headers = {'Authorization': access_token}
response = requests.get(url_users, headers=headers)
if response.status_code == 200:
users = response.json()["value"]
else:
users = []
print(response.status_code)
print("Error getting users from workspace")
users_dict_id = [principal['identifier'] for principal in users if principal.get('principalType') == 'Group']
users_dict_emailaddress = [principal['emailAddress'] for principal in users if principal.get('emailAddress', '').lower().endswith('.com')]
def grant_workspace_access(workspaceid,access_token,user,role):
url = f'https://api.powerbi.com/v1.0/myorg/groups/{workspaceid}/users'
headers = {'Authorization': access_token,'Content-Type': 'application/json'}
if '@kpmg.com' in user.lower():
payload = {
"emailAddress": user,
"groupUserAccessRight": "Member"
}
else:
if user == master_workspace_identity:
payload = {
"identifier": user,
"principalType": "App",
"groupUserAccessRight": "Admin"
}
else:
payload = {
"identifier": user,
"principalType": "Group",
"groupUserAccessRight": "Admin"
}
try:
response = requests.post(url, headers=headers, data=json.dumps(payload))
response.raise_for_status()
if response.status_code == 200:
print(f"Added '{user}'")
return 'Success'
except requests.exceptions.HTTPError as e:
if '404 Client Error' in to_str(e):
return to_str(e)
else:
rs_code=404
ct=0
while (rs_code !=200 | ct <3):
sleep(10)
response = requests.post(url, data=payload, headers=headers)
ct+=1
print(f"Failed to add '{user}'")
raise requests.HTTPError(response)
found = False
for id in admin_groups:
print(id)
for item in users_dict_id:
if id == item:
found = True
break
if found:
print(f'{id} already has access to workspace')
else:
if access_token is None or token_expiration is None or token_expiration < datetime.utcnow():
access_token, token_expiration = token_generation(client_id, client_secret, tenant_id)
message = grant_workspace_access(dest_workspaceid,access_token,id,"Admin")
found = False
for id in team_list:
print(id)
for item in users_dict_emailaddress:
if id.lower() == item.lower():
found = True
break
if found:
print(f' {id} already has access to workspace')
else:
if access_token is None or token_expiration is None or token_expiration < datetime.utcnow():
access_token, token_expiration = token_generation(client_id, client_secret, tenant_id)
message = grant_workspace_access(dest_workspaceid,access_token,id,"Member")
#############################################
@Devon_Rtk - Created post on your request
HI @kur83,
Did these content replied to the specific thread? If that is the case, you can share the original post link here and I will help you to merge them.
Regards,
Xiaoxin Sheng
Check out the September 2024 Power BI update to learn about new features.
Learn from experts, get hands-on experience, and win awesome prizes.
User | Count |
---|---|
39 | |
4 | |
3 | |
3 | |
2 |