Tutorial: Usar o SDK do Python do Microsoft Purview

Este tutorial apresentará como usar o SDK do Python do Microsoft Purview. Você pode usar o SDK para fazer todas as operações mais comuns do Microsoft Purview programaticamente, em vez de por meio do portal de governança do Microsoft Purview.

Neste tutorial, você aprenderá como usar o SDK para:

Pré-requisitos

Para este tutorial, você precisará de:

• Python 3.6 ou superior
• Uma assinatura do Azure ativa. Se não tiver uma, você pode criar uma gratuitamente.
• Um locatário do Microsoft Entra associado à sua assinatura.
• Uma conta de Armazenamento do Azure. Se ainda não tiver um, você pode seguir nosso guia de início rápido para criar um.
• Uma conta do Microsoft Purview. Se ainda não tiver um, você pode seguir nosso guia de início rápido para criar um.
• Uma entidade de serviço com um segredo de cliente.

Conceder ao Microsoft Purview acesso à conta de armazenamento

Antes de poder verificar o conteúdo da conta de armazenamento, você precisa dar ao Microsoft Purview a função correta.

1. Vá para sua conta de armazenamento por meio do portal do Azure.

2. Selecione Controle de Acesso (IAM).

3. Selecione o botão Adicionar e selecione Adicionar atribuição de função.

   

4. Na próxima janela, pesquise a função Leitor de Blob de Armazenamento e selecione-a:

   

5. Em seguida, vá para a guia Membros e selecione Selecionar membros:

   

6. Um novo painel aparece à direita. Pesquise e selecione o nome de sua instância existente do Microsoft Purview.

7. Em seguida, você pode selecionar Examinar + Atribuir.

O Microsoft Purview agora tem o direito de leitura necessário para verificar seu Armazenamento de Blobs.

Conceda ao seu aplicativo o acesso à sua conta do Microsoft Purview

1. Primeiro, você precisará da ID do Cliente, da ID do Locatário e do segredo do cliente da entidade de serviço. Para encontrar essas informações, selecione seu Microsoft Entra ID.

2. Em seguida, selecione Registros de aplicativo.

3. Selecione seu aplicativo e localize as informações necessárias:

   • Nome

   • ID do cliente (ou ID do aplicativo)

   • ID do locatário (ou ID do diretório)

     

   • Segredo do cliente

     

4. Agora você precisa fornecer as funções relevantes do Microsoft Purview à entidade de serviço. Para fazer isso, acesse sua instância do Microsoft Purview. Selecione Abrir portal de governança do Microsoft Purview ou abra o portal de governança do Microsoft Purview diretamente e escolha a instância que você implantou.

5. No portal de governança do Microsoft Purview, selecione Mapa de dados e, em seguida, Coleções:

   

6. Selecione a coleção com a qual deseja trabalhar e vá para a guia Atribuições de função . Adicione a entidade de serviço nas seguintes funções:

   • Administradores da coleção
   • Administradores da fonte de dados
   • Curadores de dados
   • Leitores de dados

7. Para cada função, selecione o botão Editar atribuições de função e selecione a função à qual você deseja adicionar a entidade de serviço. Ou selecione o botão Adicionar ao lado de cada função e adicione a entidade de serviço pesquisando seu nome ou ID do Cliente, conforme mostrado abaixo:

   

Instale os pacotes do Python

1. Abra um novo prompt de comando ou terminal
2. Instale o pacote de identidade do Azure para autenticação:

   pip install azure-identity
   

3. Instale o pacote do Cliente de Verificação do Microsoft Purview:

   pip install azure-purview-scanning
   

4. Instale o pacote do Cliente de Administração do Microsoft Purview:

   pip install azure-purview-administration
   

5. Instale o pacote do Cliente do Microsoft Purview:

   pip install azure-purview-catalog
   

6. Instale o pacote da Conta do Microsoft Purview:

   pip install azure-purview-account
   

7. Instale o pacote do Azure Core:

   pip install azure-core
   

Criar arquivo de script Python

Crie um arquivo de texto sem formatação e salve-o como um script Python com o sufixo .py. Por exemplo: tutorial.py.

Instanciar um cliente de Verificação, Catálogo e Administração

Nesta seção, você aprenderá a instanciar:

• Um cliente de verificação útil para registrar fontes de dados, criar e gerenciar regras de verificação, acionar uma verificação, etc.
• Um cliente de catálogo útil para interagir com o catálogo por meio de pesquisa, navegação pelos ativos descobertos, identificação da confidencialidade de seus dados, etc.
• Um cliente de administração é útil para interagir com o próprio Mapa de Dados do Microsoft Purview, para operações como listar coleções.

Primeiro, você precisa se autenticar no seu Microsoft Entra ID. Para isso, você usará o segredo do cliente criado.

1. Comece com as instruções de importação necessárias: nossos três clientes, a instrução de credenciais e uma instrução de exceções do Azure.

   from azure.purview.scanning import PurviewScanningClient
   from azure.purview.catalog import PurviewCatalogClient
   from azure.purview.administration.account import PurviewAccountClient
   from azure.identity import ClientSecretCredential 
   from azure.core.exceptions import HttpResponseError
   

2. Especifique as seguintes informações no código:

   • ID do cliente (ou ID do aplicativo)
   • ID do locatário (ou ID do diretório)
   • Segredo do cliente

   client_id = "<your client id>" 
   client_secret = "<your client secret>"
   tenant_id = "<your tenant id>"
   

3. Especifique o ponto de extremidade:

   Importante

   O valor do ponto de extremidade será diferente dependendo do portal do Microsoft Purview que você estiver usando. Ponto de extremidade do portal de governança do Microsoft Purview clássico: https://{your_purview_account_name}.purview.azure.com/ Ponto de extremidade do novo portal do Microsoft Purview: https://api.purview-service.microsoft.com

   Verificar ponto de extremidade para o portal de governança do Microsoft Purview clássico: https://{your_purview_account_name}.scan.purview.azure.com/ Ponto de extremidade para o novo portal do Microsoft Purview: https://api.scan.purview-service.microsoft.com

   purview_endpoint = "<endpoint>"
   
   purview_scan_endpoint = "<scan endpoint>"
   

4. Agora você pode instanciar os três clientes:

   def get_credentials():
       credentials = ClientSecretCredential(client_id=client_id, client_secret=client_secret, tenant_id=tenant_id)
       return credentials
   
   def get_purview_client():
       credentials = get_credentials()
       client = PurviewScanningClient(endpoint=purview_scan_endpoint, credential=credentials, logging_enable=True)  
       return client
   
   def get_catalog_client():
       credentials = get_credentials()
       client = PurviewCatalogClient(endpoint=purview_endpoint, credential=credentials, logging_enable=True)
       return client
   
   def get_admin_client():
       credentials = get_credentials()
       client = PurviewAccountClient(endpoint=purview_endpoint, credential=credentials, logging_enable=True)
       return client
   

Muitos de nossos scripts começarão com essas mesmas etapas, pois precisaremos desses clientes para interagir com a conta.

Registrar uma fonte de dados

Nesta seção, você registrará seu Armazenamento de Blobs.

1. Como discutimos na seção anterior, primeiro você importará os clientes necessários para acessar sua conta do Microsoft Purview. Importe também o pacote de resposta a erros do Azure para que você possa solucionar problemas e o ClientSecretCredential para construir suas credenciais do Azure.

   from azure.purview.administration.account import PurviewAccountClient
   from azure.purview.scanning import PurviewScanningClient
   from azure.core.exceptions import HttpResponseError
   from azure.identity import ClientSecretCredential
   

2. Reúna a ID do recurso para sua conta de armazenamento seguindo este guia: obtenha a ID do recurso para uma conta de armazenamento.

3. Em seguida, no arquivo Python, defina as seguintes informações para poder registrar o armazenamento de Blobs programaticamente:

   storage_name = "<name of your Storage Account>"
   storage_id = "<id of your Storage Account>"
   rg_name = "<name of your resource group>"
   rg_location = "<location of your resource group>"
   reference_name_purview = "<name of your Microsoft Purview account>"
   

4. Forneça o nome da coleção na qual você deseja registrar o armazenamento de blobs. (Deve ser a mesma coleção em que você aplicou permissões anteriormente. Se não estiver, primeiro aplique permissões a essa coleção.) Se for a coleção raiz, use o mesmo nome da instância do Microsoft Purview.

   collection_name = "<name of your collection>"
   

5. Crie uma função para construir as credenciais para acessar sua conta do Microsoft Purview:

   client_id = "<your client id>" 
   client_secret = "<your client secret>"
   tenant_id = "<your tenant id>"
   
   
   def get_credentials():
        credentials = ClientSecretCredential(client_id=client_id, client_secret=client_secret, tenant_id=tenant_id)
        return credentials
   

6. Todas as coleções no Mapa de Dados do Microsoft Purview têm um nome amigável e um nome.

   • O nome amigável é aquele que você vê na coleção. Por exemplo: Vendas.
   • O nome para todas as coleções (exceto a coleção raiz) é um nome de seis caracteres atribuído pelo mapa de dados.

   O Python precisa desse nome de seis caracteres para fazer referência a qualquer subcoleção. Para converter seu nome amigável automaticamente para o nome da coleção de seis caracteres necessário no script, adicione este bloco de código:

   Importante

   O valor do ponto de extremidade será diferente dependendo do portal do Microsoft Purview que você estiver usando. Ponto de extremidade do portal de governança do Microsoft Purview clássico: purview.azure.com/ Ponto de extremidade do novo portal do Microsoft Purview: purview.microsoft.com/

   Portanto, se você estiver usando o novo portal, o valor do seu ponto de extremidade será algo como: "https://consotopurview.scan.purview.microsoft.com"

   def get_admin_client():
        credentials = get_credentials()
        client = PurviewAccountClient(endpoint=purview_endpoint, credential=credentials, logging_enable=True)
        return client
   
   try:
     admin_client = get_admin_client()
   except ValueError as e:
       print(e)
   
   collection_list = client.collections.list_collections()
    for collection in collection_list:
     if collection["friendlyName"].lower() == collection_name.lower():
         collection_name = collection["name"]
   

7. Para ambos os clientes e, dependendo das operações, você também precisa fornecer um corpo de entrada. Para registrar uma fonte, você precisará fornecer um corpo de entrada para registro da fonte de dados:

   ds_name = "<friendly name for your data source>"
   
   body_input = {
           "kind": "AzureStorage",
           "properties": {
               "endpoint": f"https://{storage_name}.blob.core.windows.net/",
               "resourceGroup": rg_name,
               "location": rg_location,
               "resourceName": storage_name,
               "resourceId": storage_id,
               "collection": {
                   "type": "CollectionReference",
                   "referenceName": collection_name
               },
               "dataUseGovernance": "Disabled"
           }
   }    
   

8. Agora você pode chamar seus clientes do Microsoft Purview e registrar a fonte de dados.

   Importante

   O valor do ponto de extremidade será diferente dependendo do portal do Microsoft Purview que você estiver usando. Ponto de extremidade do portal de governança do Microsoft Purview clássico: https://{your_purview_account_name}.purview.azure.com/ Ponto de extremidade do novo portal do Microsoft Purview: https://api.purview-service.microsoft.com

   Se você estiver usando o portal clássico, o valor do ponto de extremidade será: https://{your_purview_account_name}.scan.purview.azure.com Se você estiver usando o novo portal, o valor do ponto de extremidade será: https://scan.api.purview-service.microsoft.com

   def get_purview_client():
        credentials = get_credentials()
        client = PurviewScanningClient(endpoint={{ENDPOINT}}, credential=credentials, logging_enable=True)  
        return client
   
   try:
       client = get_purview_client()
   except ValueError as e:
       print(e)
   
   try:
       response = client.data_sources.create_or_update(ds_name, body=body_input)
       print(response)
       print(f"Data source {ds_name} successfully created or updated")
   except HttpResponseError as e:
       print(e)
   

Quando o processo de registro é bem-sucedido, você pode ver uma resposta corporal enriquecida do cliente.

Nas seções a seguir, você examinará a fonte de dados registrada e pesquisará o catálogo. Cada um desses scripts será estruturado de forma semelhante a este script de registro.

Código completo

from azure.purview.scanning import PurviewScanningClient
from azure.identity import ClientSecretCredential 
from azure.core.exceptions import HttpResponseError
from azure.purview.administration.account import PurviewAccountClient

client_id = "<your client id>" 
client_secret = "<your client secret>"
tenant_id = "<your tenant id>"
purview_endpoint = "<endpoint>"
purview_scan_endpoint = "<scan endpoint>"
storage_name = "<name of your Storage Account>"
storage_id = "<id of your Storage Account>"
rg_name = "<name of your resource group>"
rg_location = "<location of your resource group>"
collection_name = "<name of your collection>"
ds_name = "<friendly data source name>"

def get_credentials():
	credentials = ClientSecretCredential(client_id=client_id, client_secret=client_secret, tenant_id=tenant_id)
	return credentials

def get_purview_client():
	credentials = get_credentials()
	client = PurviewScanningClient(endpoint=purview_scan_endpoint, credential=credentials, logging_enable=True)  
	return client

def get_admin_client():
	credentials = get_credentials()
	client = PurviewAccountClient(endpoint=purview_endpoint, credential=credentials, logging_enable=True)
	return client

try:
	admin_client = get_admin_client()
except ValueError as e:
        print(e)

collection_list = admin_client.collections.list_collections()
for collection in collection_list:
	if collection["friendlyName"].lower() == collection_name.lower():
		collection_name = collection["name"]


body_input = {
	"kind": "AzureStorage",
	"properties": {
		"endpoint": f"https://{storage_name}.blob.core.windows.net/",
		"resourceGroup": rg_name,
		"location": rg_location,
		"resourceName": storage_name,
 		"resourceId": storage_id,
		"collection": {
			"type": "CollectionReference",
			"referenceName": collection_name
		},
		"dataUseGovernance": "Disabled"
	}
}

try:
	client = get_purview_client()
except ValueError as e:
        print(e)

try:
	response = client.data_sources.create_or_update(ds_name, body=body_input)
	print(response)
	print(f"Data source {ds_name} successfully created or updated")
except HttpResponseError as e:
    print(e)

Verificar a fonte de dados

A verificação de uma fonte de dados pode ser feita em duas etapas:

1. Criar uma definição de verificação
2. Disparar uma execução de verificação

Neste tutorial, você usará as regras de verificação padrão para contêineres do Armazenamento de Blobs. No entanto, você também pode criar regras de verificação personalizadas programaticamente com o Cliente de Verificação do Microsoft Purview.

Agora vamos verificar a fonte de dados registrada acima.

1. Adicione uma instrução de importação para gerar um identificador exclusivo, chame o cliente de verificação do Microsoft Purview, o cliente de administração do Microsoft Purview, o pacote de resposta a erros do Azure para poder solucionar problemas e a credencial secreta do cliente para coletar suas credenciais do Azure.

   import uuid
   from azure.purview.scanning import PurviewScanningClient
   from azure.purview.administration.account import PurviewAccountClient
   from azure.core.exceptions import HttpResponseError
   from azure.identity import ClientSecretCredential 
   

2. Crie um cliente de verificação usando suas credenciais:

   client_id = "<your client id>" 
   client_secret = "<your client secret>"
   tenant_id = "<your tenant id>"
   
   def get_credentials():
        credentials = ClientSecretCredential(client_id=client_id, client_secret=client_secret, tenant_id=tenant_id)
        return credentials
   
   def get_purview_client():
        credentials = get_credentials()
        client = PurviewScanningClient(endpoint=f"https://{reference_name_purview}.scan.purview.azure.com", credential=credentials, logging_enable=True)  
        return client
   
   try:
        client = get_purview_client()
   except ValueError as e:
        print(e)
   

3. Adicione o código para coletar o nome interno da coleção. (Para obter mais informações, consulte a seção anterior):

   collection_name = "<name of the collection where you will be creating the scan>"
   
   def get_admin_client():
        credentials = get_credentials()
        client = PurviewAccountClient(endpoint=purview_endpoint, credential=credentials, logging_enable=True)
        return client
   
   try:
       admin_client = get_admin_client()
   except ValueError as e:
       print(e)
   
   collection_list = client.collections.list_collections()
    for collection in collection_list:
     if collection["friendlyName"].lower() == collection_name.lower():
         collection_name = collection["name"]
   

4. Em seguida, crie uma definição de verificação:

   ds_name = "<name of your registered data source>"
   scan_name = "<name of the scan you want to define>"
   reference_name_purview = "<name of your Microsoft Purview account>"
   
   body_input = {
           "kind":"AzureStorageMsi",
           "properties": { 
               "scanRulesetName": "AzureStorage", 
               "scanRulesetType": "System", #We use the default scan rule set 
               "collection": 
                   {
                       "referenceName": collection_name,
                       "type": "CollectionReference"
                   }
           }
   }
   
   try:
       response = client.scans.create_or_update(data_source_name=ds_name, scan_name=scan_name, body=body_input)
       print(response)
       print(f"Scan {scan_name} successfully created or updated")
   except HttpResponseError as e:
       print(e)
   

5. Agora que a verificação está definida, você pode disparar uma execução de verificação com uma ID exclusiva:

   run_id = uuid.uuid4() #unique id of the new scan
   
   try:
       response = client.scan_result.run_scan(data_source_name=ds_name, scan_name=scan_name, run_id=run_id)
       print(response)
       print(f"Scan {scan_name} successfully started")
   except HttpResponseError as e:
       print(e)
   

Código completo

import uuid
from azure.purview.scanning import PurviewScanningClient
from azure.purview.administration.account import PurviewAccountClient
from azure.identity import ClientSecretCredential

ds_name = "<name of your registered data source>"
scan_name = "<name of the scan you want to define>"
reference_name_purview = "<name of your Microsoft Purview account>"
client_id = "<your client id>" 
client_secret = "<your client secret>"
tenant_id = "<your tenant id>"
collection_name = "<name of the collection where you will be creating the scan>"

def get_credentials():
	credentials = ClientSecretCredential(client_id=client_id, client_secret=client_secret, tenant_id=tenant_id)
	return credentials

def get_purview_client():
	credentials = get_credentials()
	client = PurviewScanningClient(endpoint=purview_scan_endpoint, credential=credentials, logging_enable=True)  
	return client

def get_admin_client():
	credentials = get_credentials()
	client = PurviewAccountClient(endpoint=purview_endpoint, credential=credentials, logging_enable=True)
	return client

try:
	admin_client = get_admin_client()
except ValueError as e:
        print(e)

collection_list = admin_client.collections.list_collections()
for collection in collection_list:
	if collection["friendlyName"].lower() == collection_name.lower():
		collection_name = collection["name"]


try:
	client = get_purview_client()
except AzureError as e:
	print(e)

body_input = {
	"kind":"AzureStorageMsi",
	"properties": { 
		"scanRulesetName": "AzureStorage", 
		"scanRulesetType": "System",
		"collection": {
			"type": "CollectionReference",
			"referenceName": collection_name
		}
	}
}

try:
	response = client.scans.create_or_update(data_source_name=ds_name, scan_name=scan_name, body=body_input)
	print(response)
	print(f"Scan {scan_name} successfully created or updated")
except HttpResponseError as e:
	print(e)

run_id = uuid.uuid4() #unique id of the new scan

try:
	response = client.scan_result.run_scan(data_source_name=ds_name, scan_name=scan_name, run_id=run_id)
	print(response)
	print(f"Scan {scan_name} successfully started")
except HttpResponseError as e:
	print(e)

Pesquisar catálogo

Depois que uma verificação é concluída, é provável que os ativos tenham sido descobertos e até mesmo classificados. Esse processo pode levar algum tempo para ser concluído após uma verificação, portanto, talvez seja necessário aguardar antes de executar a próxima parte do código. Aguarde até que a verificação seja concluída e os ativos apareçam no Catálogo unificado do Microsoft Purview.

Depois que os ativos estiverem prontos, você poderá usar o cliente do Catálogo do Microsoft Purview para pesquisar todo o catálogo.

1. Desta vez, você precisa importar o cliente de catálogo em vez do de digitalização. Inclua também o erro HTTPResponse e ClientSecretCredential.

   from azure.purview.catalog import PurviewCatalogClient
   from azure.identity import ClientSecretCredential 
   from azure.core.exceptions import HttpResponseError
   

2. Crie uma função para obter as credenciais para acessar sua conta do Microsoft Purview e instancie o cliente do catálogo.

   client_id = "<your client id>" 
   client_secret = "<your client secret>"
   tenant_id = "<your tenant id>"
   reference_name_purview = "<name of your Microsoft Purview account>"
   
   def get_credentials():
        credentials = ClientSecretCredential(client_id=client_id, client_secret=client_secret, tenant_id=tenant_id)
        return credentials
   
   def get_catalog_client():
       credentials = get_credentials()
       client = PurviewCatalogClient(endpoint=f"https://{reference_name_purview}.scan.purview.azure.com", credential=credentials, logging_enable=True)
       return client
   
   try:
       client_catalog = get_catalog_client()
   except ValueError as e:
       print(e)  
   

3. Configure os critérios de pesquisa e as palavras-chave no corpo de entrada:

   keywords = "keywords you want to search"
   
   body_input={
       "keywords": keywords
   }
   

   Aqui você especifica apenas palavras-chave, mas lembre-se de que pode adicionar muitos outros campos para especificar ainda mais sua consulta.

4. Pesquise no catálogo:

   try:
       response = client_catalog.discovery.query(search_request=body_input)
       print(response)
   except HttpResponseError as e:
       print(e)
   

Código completo

from azure.purview.catalog import PurviewCatalogClient
from azure.identity import ClientSecretCredential 
from azure.core.exceptions import HttpResponseError

client_id = "<your client id>" 
client_secret = "<your client secret>"
tenant_id = "<your tenant id>"
reference_name_purview = "<name of your Microsoft Purview account>"
keywords = "<keywords you want to search for>"

def get_credentials():
	credentials = ClientSecretCredential(client_id=client_id, client_secret=client_secret, tenant_id=tenant_id)
	return credentials

def get_catalog_client():
	credentials = get_credentials()
	client = PurviewCatalogClient(endpoint=purview_endpoint, credential=credentials, logging_enable=True)
	return client

body_input={
	"keywords": keywords
}

try:
	catalog_client = get_catalog_client()
except ValueError as e:
	print(e)

try:
	response = catalog_client.discovery.query(search_request=body_input)
	print(response)
except HttpResponseError as e:
	print(e)

Excluir uma fonte de dados

Nesta seção, você aprenderá a excluir a fonte de dados registrada anteriormente. Esta operação é bastante simples e é feita com o cliente de digitalização.

1. Importar o cliente de digitalização . Inclua também o erro HTTPResponse e ClientSecretCredential.

   from azure.purview.scanning import PurviewScanningClient
   from azure.identity import ClientSecretCredential 
   from azure.core.exceptions import HttpResponseError
   

2. Crie uma função para obter as credenciais para acessar sua conta do Microsoft Purview e instancie o cliente de verificação.

   client_id = "<your client id>" 
   client_secret = "<your client secret>"
   tenant_id = "<your tenant id>"
   reference_name_purview = "<name of your Microsoft Purview account>"
   
   def get_credentials():
        credentials = ClientSecretCredential(client_id=client_id, client_secret=client_secret, tenant_id=tenant_id)
        return credentials
   
   def get_scanning_client():
       credentials = get_credentials()
       PurviewScanningClient(endpoint=f"https://{reference_name_purview}.scan.purview.azure.com", credential=credentials, logging_enable=True) 
       return client
   
   try:
       client_scanning = get_scanning_client()
   except ValueError as e:
       print(e)  
   

3. Excluir a fonte de dados:

       ds_name = "<name of the registered data source you want to delete>"
       try:
           response = client_scanning.data_sources.delete(ds_name)
           print(response)
           print(f"Data source {ds_name} successfully deleted")
       except HttpResponseError as e:
           print(e)
   

Código completo

from azure.purview.scanning import PurviewScanningClient
from azure.identity import ClientSecretCredential 
from azure.core.exceptions import HttpResponseError


client_id = "<your client id>" 
client_secret = "<your client secret>"
tenant_id = "<your tenant id>"
reference_name_purview = "<name of your Microsoft Purview account>"
ds_name = "<name of the registered data source you want to delete>"

def get_credentials():
	credentials = ClientSecretCredential(client_id=client_id, client_secret=client_secret, tenant_id=tenant_id)
	return credentials

def get_scanning_client():
	credentials = get_credentials()
	client = PurviewScanningClient(endpoint=f"https://{reference_name_purview}.scan.purview.azure.com", credential=credentials, logging_enable=True) 
	return client

try:
	client_scanning = get_scanning_client()
except ValueError as e:
	print(e)  

try:
	response = client_scanning.data_sources.delete(ds_name)
	print(response)
	print(f"Data source {ds_name} successfully deleted")
except HttpResponseError as e:
	print(e)