Create Purview Lineage via Python

CreatePurviewEntityLineage.py

from pyapacheatlas.auth import ServicePrincipalAuthentication
from pyapacheatlas.auth import BasicAuthentication
from pyapacheatlas.core import PurviewClient, AtlasEntity, AtlasProcess
from pyapacheatlas.core.util import AtlasException, GuidTracker
from json import dumps

# Configure Service Principal Authentication
auth = ServicePrincipalAuthentication(
    tenant_id = "xxxxxxx-xxxxxx-xxxxx-xxxxx-xxxxxxxxxx", 
    client_id = "xxxxxx-xxxxx-xxxxx-xxxxx-xxxxxxx", # Client ID, 
    client_secret = dbutils.secrets.get(scope = "azurekeyvault_secrets", key = "client-secret") # Get secret from Key Vault
)

# Instantiate Purview Client
client = PurviewClient(
    account_name = "deltalakepoc",
    authentication = auth
)

try:
    # Identity of the 'source'
    # TODO: Edit entity object to match IDs in Purview
    input01 = AtlasEntity(
        name="test",
        typeName="DataSet",
        qualified_name=https://[NAME].dfs.core.windows.net/test,
        guid="-100"
    )
    # Identity of the 'target'
    # TODO: Edit entity object to match IDs in Purview
    output01 = AtlasEntity(
        name="test-container",
        typeName="DataSet",
        qualified_name=https://[NAME].dfs.core.windows.net/test-container,
        guid="-101"
    )
    
    # Lineage component connecting 'inputs' to 'outputs'
    process = AtlasProcess(
        name="Lineage Component Name",
        typeName="Process",
        qualified_name="pyapacheatlas://democustomprocess",
        inputs=[input01],
        outputs=[output01],
        guid="-102"
    )
  
    # Convert entities into json before uploading
    results = client.upload_entities(
        batch=[output01, input01, process]
    )
    
    print(dumps(results, indent=2))
except AtlasException as e:
    print("The provided classification was not found on the provided entity.")
    print(e)
except Exception as e:
    print (e)