🌐

Quick Start: Reusable Bulk Upload Python Script

Get GUID from qualified name β†’ Updated different metadata using the GUID

import requests
import pandas as pd
import time
import json
import numpy as np

BASE_URL = 'https://client.atlan.com/'
HEADERS = {
    'Content-Type': 'application/json',
    'APIKEY': '<api key>'
    }


def find_guid(qualifiedName):
    url = BASE_URL + "api/metadata/atlas/tenants/default/search/basic"
    payload = {
            "searchType": "BASIC",
            "typeName": "AtlanAsset",
            "excludeDeletedEntities": True,
            "includeClassificationAttributes": False,
            "includeSubClassifications": False,
            "includeSubTypes": True,
            "limit": 300,
            "offset": 0,
            "attributes": [
                "guid"
            ],
            "minScore": 100,
            "query": "",
            "entityFilters": {
                "condition": "AND",
                "criterion": [
                    {
                        "condition": "OR",
                        "criterion": [
                            {
                                "attributeName": "qualifiedName",
                                "attributeValue": qualifiedName,
                                "operator": "eq"
                            }
                        ]
                    }
                ]
            }
        }
    response = requests.request("POST", url, headers=HEADERS, json=payload)
    if response.status_code != 200:
        print(response)
        print(payload)
        time.sleep(500)
        response.raise_for_status
    else:
        res = json.loads(response.text)
        # print(res)
        # time.slee
        entities = res.get('entities')
        guid = entities[0].get('guid')
        return guid

def update_description():
    df = pd.read_csv('client_columns.csv')
    for i, row in df.iterrows():
        # print('\n')
        if isinstance(row['asset_guid'], float):
            # print(row)
            qf_name = row['qualifiedName']
            if isinstance(row['qualifiedName'], float):
                print('making qualified name')
                qf_name = row['data_source'] + row['host'] + row['database'] + "/" + row['schema'] + '/' + row['table_name'] + '/' + row['column_name']
                print(qf_name)
            guid = find_guid(qf_name)
            print(guid)
            df.at[i, 'asset_guid'] = guid
        if isinstance(row['updated_description'], float):
            continue
        df.to_csv('assets_with_guid.csv')
        if row['updated_description'] == '':
            continue
        url = BASE_URL + 'api/metadata/atlas/tenants/default/entity/guid/' + row['asset_guid'] + "?name=customDescription"
        if '"' in row['updated_description']:
            df.at[i, 'updated_description'] = row['updated_description'].replace('"', '''\\\"''')
        payload="\"{0}\"".format(row['updated_description'])
        response = requests.request("PUT", url, headers=HEADERS, data=payload)
        print(url)
        if response.status_code != 200:
            print('Not able to update for guid : {guid}'.format(guid=row['asset_guid']))
            print(response.text)
            print(response)
            response.raise_for_status()
            time.sleep(20)
        else:
            print('Updated description for guid: {guid}'.format(guid=row['asset_guid'])) 
            time.sleep(0.3)


def update_status():
    df = pd.read_csv('client_columns.csv')
    for i, row in df.iterrows():
        print('\n')
        url = BASE_URL + 'api/metadata/atlas/tenants/default/entity/guid/' + row['asset_guid'] + "?name=status"
        payload="\"{0}\"".format(row['updated_status'])
        response = requests.request("PUT", url, headers=HEADERS, data=payload)
        print(url)
        if response.status_code != 200:
            print('Not able to status for guid : {guid}'.format(guid=row['asset_guid']))
            print(response.text)
        else:
            print('Updated status for guid: {guid}'.format(guid=row['asset_guid'])) 
            time.sleep(0.1)

def main():
    update_description()
    update_status()

if __name__ == "__main__":
    main()