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
