create-collections.py

import requests
import json
from termcolor import colored

def aadtoken(client_id, client_secret, client_tenant):
    url = "https://login.microsoftonline.com/{}/oauth2/token".format(client_tenant)
    payload='grant_type=client_credentials&client_id={}&client_secret={}&resource=https%3A%2F%2Fpurview.azure.net'.format(client_id, client_secret)
    response = requests.request("POST", url, data=payload)

    return json.loads(response.text)['access_token']

def get_collection(access_token, collections_name):
    url = "https://{}.purview.azure.com/collections/{}?api-version=2019-11-01-preview".format(purview_account, collections_name)
    headers = {
        'Authorization': 'Bearer {}'.format(access_token),
    }
    response = requests.request("GET", url, headers=headers)
    return response.text

def delete_collection(access_token, collections_name):
    url = "https://{}.purview.azure.com/collections/{}?api-version=2019-11-01-preview".format(purview_account, collections_name)
    headers = {
        'Authorization': 'Bearer {}'.format(access_token),
    }
    response = requests.request("DELETE", url, headers=headers)
    return response.text

def create_collection(access_token, collections_name, collections_friendly_name, parent_collection_name):
    url = "https://{}.purview.azure.com/collections/{}?api-version=2019-11-01-preview".format(purview_account, collections_name)
    headers = {
        'Authorization': 'Bearer {}'.format(access_token),
        'Content-Type': 'application/json'
    }
    payload = json.dumps({
                "name": collections_name,
                "parentCollection": {
                    "type": "CollectionReference",
                    "referenceName": parent_collection_name
                },
                "friendlyName": collections_friendly_name
            })
    response = requests.request("PUT", url, headers=headers, data=payload)
    return response.text

def customprint(response):
    # If response contains "unauthorized" - print
    if "Unauthorized" in response:
        print(colored(response, 'red'))
        # raise Exception('Hit the authorization error!')
    else:
        print(colored(response, 'green'))


### MAIN LOOP ###
def runloop(access_token, purview_account):
    for i in range(1, 1000000):
        print('Round: {}'.format(i))

        print('\n Creating collections: \n')

        for j in range(1, 5):
            customprint(create_collection(access_token, "ADLSGen2", "ADLSGen2", "Bento")) # Level 3    
            customprint(create_collection(access_token, "Bento", "Bento", "TS03")) # Level 2
            customprint(create_collection(access_token, "TS03", "TS03", purview_account)) # Level 1
            customprint(create_collection(access_token, "Bento", "Bento", "TS03")) # Level 2
            customprint(create_collection(access_token, "ADLSGen2", "ADLSGen2", "Bento")) # Level 3

        print('\n Get collections: \n')

        for j in range(1, 5):
            customprint(get_collection(access_token, "TS03"))
            customprint(get_collection(access_token, "Bento"))
            customprint(get_collection(access_token, "ADLSGen2"))

        print('\n Delete collections: \n')

        for j in range(1, 5):
            customprint(delete_collection(access_token, "ADLSGen2"))
            customprint(delete_collection(access_token, "Bento"))
            customprint(delete_collection(access_token, "TS03"))

        print('\n\n\n')
    
    return null

if __name__ == "__main__":
    client_id = 'your-client-id'
    client_secret = 'your-client-secret'
    client_tenant = 'your-tenant-id'
    purview_account = 'your-pvw-acct'

    access_token = aadtoken(client_id, client_secret, client_tenant)
    
    try:
        runloop(access_token, purview_account)
    except Exception:
        pass