Need help to create Lakehouse programmatically using Service Principal

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