autotagger.ipynb

{
    "cells": [
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "<h3>Auto-tagger v1</h3>\n",
                "\n",
                "This script attempts to iterate through a list of ADLS storage assets and apply (managed) attributes against each asset in order to make them more discoverable and improve the search ranking when adding attribute terms to search term.\n",
                "The business logic iterates through assets and extracts folder names to a pre-defined depth in the folder hierarchy. Folder names can then be compared against your own key value pairs, to extract the business value. This is a common approach for a data lake structure based on a predefined taxonomy where folder names are acronyms (the keys).\n",
                "Currently this script only supports ADLS assets and two folder levels deep, in this example which is client and market.\n",
                "\n",
                "<small>MIT License\n",
                "\n",
                "Copyright (c) 2022 Nick Hurt\n",
                "\n",
                "Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the \"Software\"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom  he Software is furnished to do so, subject to the following conditions:  The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.\n",
                "\n",
                "THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.</small>"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "<h4>Option 1: Load lookup values from json file into dictionary object</h4>"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "import json\n",
                "with open(\"clients.json\", encoding='utf-8') as f:\n",
                "  #format should be {\"code\":\"A_AAIR\",\"name\":\"AA Ireland\"},{\"code\":\"A_ABBO\",\"name\":\"Abbott\"},{\"code\":\"A_ACFH\",\"name\":\"Abercrombie and Fitch\"}...{}\n",
                "  clientjson = json.load(f)\n",
                "clientdic = dict()\n",
                "for client in clientjson:\n",
                "    #print(client)\n",
                "    clientdic[client['code']] = client['name']\n",
                "\n",
                "#print(list(clientdic))\n",
                "\n",
                "\n",
                "with open(\"markets.json\", encoding='utf-8') as f:\n",
                "  #format should be {\"aliasName\":\"Afghanistan\",\"code\":\"AF\",\"threeLetterCode\":\"AFG\",\"name\":\"Afghanistan\"},{\"aliasName\":\"Albania\",\"code\":\"AL\",\"threeLetterCode\":\"ALB\",\"name\":\"Albania\"},...{}\n",
                "  marketjson = json.load(f)\n",
                "marketdic = dict()\n",
                "for market in marketjson:\n",
                "    #print(client)\n",
                "    marketdic[market['code']] = market['name']\n",
                "\n",
                "#print(list(marketdic))"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "<h4>Option 2: Load lookup values from SQL database table(s) into dictionary object</h4>\n",
                "Simple boilerplate code to populate a dictionary object from two columns from SQL table.  Replace the [tokens] with your details"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "import pyodbc\n",
                "# Replace the [tokens] with your details\n",
                "cnxn = pyodbc.connect(\"Driver={ODBC Driver 17 for SQL Server};Server=tcp:[azuresqlsrvname].database.windows.net,1433;Database=[databasename]];Uid=[username];Pwd=[password];Encrypt=yes;TrustServerCertificate=no;Connection Timeout=10;\")\n",
                "cursor = cnxn.cursor()\n",
                "clientdic = dict()\n",
                "\n",
                "try:\n",
                "    fetchsql = \"\"\"select client_key, client_value from lookuptable \"\"\"\n",
                "    cursor.execute(fetchsql)\n",
                "    row = cursor.fetchone()\n",
                "    while row:\n",
                "         clientdic[row[0]] = row[1]\n",
                "         row = cursor.fetchone()\n",
                "\n",
                "except pyodbc.DatabaseError as err:\n",
                "    #cnxn.commit()\n",
                "    sqlstate = err.args[1]\n",
                "    sqlstate = sqlstate.split(\".\")\n",
                "    print('Warning occured while trying to populate diction from sql table. Error message: '.join(sqlstate))\n",
                "    \n"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "<h4> Obtain auth token to interact with the Purview (Atlas) APIs</h4>\n",
                "Fill the variable values below.\n",
                "Requires service principal and secret."
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {
                "azdata_cell_guid": "e9864a7e-f566-425c-ab72-d38904d6e4b2",
                "tags": [
                    "hide_input"
                ]
            },
            "outputs": [],
            "source": [
                "#pip install requests\n",
                "\n",
                "import requests\n",
                "import json\n",
                "# Client credentials + Purview details (See https://docs.microsoft.com/en-us/azure/purview/tutorial-using-rest-apis)\n",
                "\n",
                "purview_account_name = \"\"\n",
                "client_id = \"\"\n",
                "client_secret = \"\"\n",
                "resource = \"https://purview.azure.net\"\n",
                "tenant_id = \"\"\n",
                "\n",
                "\n",
                "# oauth2 login\n",
                "url = \"https://login.microsoftonline.com/\" + tenant_id + \"/oauth2/token\"\n",
                "\n",
                "# Login and get token\n",
                "payload='grant_type=client_credentials&client_id=' + client_id + '&client_secret=' + client_secret + '&resource=' + resource\n",
                "headers = {\n",
                "  'Content-Type': 'application/x-www-form-urlencoded',\n",
                "}\n",
                "\n",
                "response = json.loads(requests.request(\"POST\", url, headers=headers, data=payload).content)\n",
                "\n",
                "# Set the access token\n",
                "access_token = response['access_token']\n"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "<h4>Create the attribute definition</h4>\n",
                "See https://docs.microsoft.com/en-us/azure/purview/tutorial-atlas-2-2-apis"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "purview_endpoint = \"https://\"+purview_account_name+\".purview.azure.com\" \n",
                "headers = {\n",
                "  'Authorization': 'Bearer ' + access_token,\n",
                "  'Content-Type': 'application/json'\n",
                "}\n",
                "tagName=''\n",
                "if tagName == '':\n",
                "  url = purview_endpoint + \"/catalog/api/atlas/v2/types/typedefs\"\n",
                "  print(url)\n",
                "  body = json.dumps({\"businessMetadataDefs\":[{\"category\":\"BUSINESS_METADATA\",\"createdBy\":\"admin\",\"updatedBy\":\"admin\",\"version\":1,\"typeVersion\":\"1.1\",\"name\":\"Data Product\",\"description\":\"\",\"attributeDefs\":[{\"name\":\"Client\",\"typeName\":\"string\",\"isOptional\":True,\"cardinality\":\"SINGLE\",\"isUnique\":False,\"isIndexable\":True,\"options\":{\"maxStrLength\":\"500\",\"applicableEntityTypes\":\"[\\\"Referenceable\\\"]\"}},{\"name\":\"Market\",\"typeName\":\"string\",\"isOptional\":True,\"cardinality\":\"SINGLE\",\"isUnique\":False,\"isIndexable\":True,\"options\":{\"maxStrLength\":\"500\",\"applicableEntityTypes\":\"[\\\"Referenceable\\\"]\"}}]}]})\n",
                "\n",
                "print(body)\n",
                "response = json.loads(requests.request(\"POST\", url, headers=headers, data=body).content)\n",
                "print(response)"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "<h4>Search for assets to apply (managed) attributes against.</h4> \n",
                "Currently there is no way to do this in bulk so it may take a significant amount of time depending on the number of assets.\n",
                "Note there is also a limit of 100,000 assets that can be returned from a search so it may be best to try to do this per collection.\n",
                "If the catalog contains more than 100,000 assets (excluding glossary terms), then the search is performed in federated manner using \"facets\". \"facet_type\" is configurable. Based on the best possible distrubtion of assets (to be under 100,000 by sub type) choose the facet_type (e.g. entityType, assetTpye, collectionId). If the sub-types still have more than 100,000 assets then only upto 100,000 assets will be exported for those sub types."
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "purview_endpoint = \"https://\"+purview_account_name+\".catalog.purview.azure.com\" \n",
                "url = purview_endpoint + \"/api/search/query?api-version=2021-05-01-preview\"\n",
                "\n",
                "# Search for assets to apply managed attributes to. In this example we search for all resource sets under the gold directory within a particular collection. \n",
                "# Apply additional filters as necessary, here are some examples to add to the payload.\n",
                "# \n",
                "# eg \"filter\": {\"entityType\": \"azure_datalake_gen2_resource_set\"},\n",
                "#    \"filter\": {\"entityType\": \"azure_datalake_gen2_path\"},\n",
                "#    \"filter\": {\"collectionId\": \"myCollectionName\"},\n",
                "payload = json.dumps({\n",
                "  \"keywords\": \"https://somedatalake.dfs.core.windows.net/gold/*\",\n",
                "  \"filter\":{\"collectionId\": \"PurviewDemo1\",\"entityType\": \"azure_datalake_gen2_resource_set\"},\n",
                "  \"limit\": 100\n",
                "})\n",
                "headers = {\n",
                "  'Authorization': 'Bearer ' + access_token,\n",
                "  'Content-Type': 'application/json'\n",
                "}\n",
                "\n",
                "response = json.loads(requests.request(\"POST\", url, headers=headers, data=payload).content)\n",
                "print('number of assets found: ' + str(response['@search.count']))\n",
                "noassets = len(response['value'])\n",
                "#print(noassets)\n",
                "\n",
                "purviewresults = response['value']\n",
                "#print(purviewresults)\n"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "<h4>Iterate through the results and depending on the folder depth will use different lookup values</h4>\n"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "for a in range(0,noassets):\n",
                "  adlpath = purviewresults[a]['qualifiedName']\n",
                "  #print(adlpath)\n",
                "  pathstr = adlpath.replace(\"https://\",\"\")\n",
                "  pathparts = pathstr.split(\"/\",-1)\n",
                "  i=0\n",
                "  tagName=''\n",
                "  for folder in pathparts:\n",
                "    if i!=0 and i<=2:\n",
                "      #print(\"level \" + str(i) + \" token : \" +folder)\n",
                "      if i == 1: # client lookup\n",
                "        # perform any lightweight cleansing/data corrections\n",
                "        cleanfolder = folder.upper()#.replace('-','_').replace('OUR','OURW')\n",
                "        try:\n",
                "          clienttag = clientdic[cleanfolder]\n",
                "        except KeyError:\n",
                "          clienttag = ''\n",
                "\n",
                "      if i == 2: # market look up\n",
                "        # perform any lightweight cleansing/data corrections\n",
                "        cleanfolder = folder.upper()#.replace('UK','GB').replace(\"’\",\"'\")\n",
                "        try:\n",
                "          markettag = marketdic[cleanfolder]\n",
                "        except KeyError:\n",
                "          markettag = ''\n",
                "\n",
                "    i=i+1\n",
                "  entityGuid = purviewresults[a]['id']  \n",
                "  print(entityGuid + ' - ' + clienttag + ' - ' + markettag)\n",
                "\n",
                "  url = purview_endpoint + \"/catalog/api/atlas/v2/entity/guid/\"+entityGuid + \"/businessmetadata?isOverwrite=true\"\n",
                "\n",
                "  headers = {\n",
                "    'Authorization': 'Bearer ' + access_token,\n",
                "    'Content-Type': 'application/json'\n",
                "  }\n",
                "  tagName=''\n",
                "  if clienttag != '' or markettag != '':\n",
                "    body = json.dumps(\n",
                "      {\n",
                "    \"Data Product\": {\n",
                "          \"Client\": clienttag,\n",
                "          \"Market\": markettag\n",
                "        }\n",
                "      }\n",
                "    )\n",
                "  response = requests.request(\"POST\", url, headers=headers, data=body)\n",
                "  print(response.text)\n",
                "print('done')\n"
            ]
        }
    ],
    "metadata": {
        "interpreter": {
            "hash": "9650cb4e16cdd4a8e8e2d128bf38d875813998db22a3c986335f89e0cb4d7bb2"
        },
        "kernelspec": {
            "display_name": "Python 3.8.3 64-bit",
            "language": "python",
            "name": "python3"
        },
        "language_info": {
            "codemirror_mode": {
                "name": "ipython",
                "version": 3
            },
            "file_extension": ".py",
            "mimetype": "text/x-python",
            "name": "python",
            "nbconvert_exporter": "python",
            "pygments_lexer": "ipython3",
            "version": "3.8.3"
        }
    },
    "nbformat": 4,
    "nbformat_minor": 2
}