Skip to main content
cancel
Showing results for 
Search instead for 
Did you mean: 

Find everything you need to get certified on Fabric—skills challenges, live sessions, exam prep, role guidance, and more. Get started

Reply
kur83
Frequent Visitor

Workspace Creation, capacity assignment, Lakehouse creation and User access - Rest API Python

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

1 REPLY 1
v-shex-msft
Community Support
Community Support

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

Community Support Team _ Xiaoxin
If this post helps, please consider accept as solution to help other members find it more quickly.

Helpful resources

Announcements
Sept PBI Carousel

Power BI Monthly Update - September 2024

Check out the September 2024 Power BI update to learn about new features.

September Hackathon Carousel

Microsoft Fabric & AI Learning Hackathon

Learn from experts, get hands-on experience, and win awesome prizes.

Sept NL Carousel

Fabric Community Update - September 2024

Find out what's new and trending in the Fabric Community.

Top Solution Authors