Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment