Tutorial: Utilizar o SDK Python do Microsoft Purview
Este tutorial irá apresentar-lhe a utilização do SDK Python do Microsoft Purview. Pode utilizar o SDK para realizar todas as operações mais comuns do Microsoft Purview programaticamente, em vez de através do portal de governação do Microsoft Purview.
Neste tutorial, irá aprender como o SDK:
- Conceder os direitos necessários para trabalhar programaticamente com o Microsoft Purview
- Registar um contentor de Armazenamento de Blobs como uma origem de dados no Microsoft Purview
- Definir e executar uma análise
- Procurar no catálogo
- Eliminar uma origem de dados
Pré-requisitos
Para este tutorial, irá precisar de:
- Python 3.6 ou superior
- Uma Subscrição do Azure ativa. Se não tiver uma, pode criar uma gratuitamente.
- Um inquilino Microsoft Entra associado à sua subscrição.
- Uma conta de Armazenamento do Azure. Se ainda não tiver uma, pode seguir o nosso guia de início rápido para criar um.
- Uma conta do Microsoft Purview. Se ainda não tiver uma, pode seguir o nosso guia de início rápido para criar um.
- Um principal de serviço com um segredo do cliente.
Importante
Para estes scripts, o valor do ponto final será diferente consoante o portal do Microsoft Purview que estiver a utilizar. Ponto final do portal de governação clássico do Microsoft Purview: purview.azure.com/ Ponto Final para o novo portal do Microsoft Purview: purview.microsoft.com/
Por isso, se estiver a utilizar o novo portal, o valor do ponto final será semelhante a: "https://consotopurview.scan.purview.microsoft.com"
Conceder acesso ao Microsoft Purview à conta de Armazenamento
Antes de poder analisar o conteúdo da conta de Armazenamento, tem de atribuir ao Microsoft Purview a função certa.
Aceda à Sua Conta de Armazenamento através do portal do Azure.
Selecione Controle de Acesso (IAM).
Selecione o botão Adicionar e selecione Adicionar atribuição de função.
Na janela seguinte, procure a função Leitor de Blobs de Armazenamento e selecione-a:
Em seguida, aceda ao separador Membros e selecione Selecionar membros:
É apresentado um novo painel à direita. Procure e selecione o nome da sua instância do Microsoft Purview existente.
Em seguida, pode selecionar Rever + Atribuir.
O Microsoft Purview tem agora o direito de leitura necessário para analisar o Armazenamento de Blobs.
Conceder à sua aplicação o acesso à sua conta do Microsoft Purview
Em primeiro lugar, precisará do ID do Cliente, do ID do Inquilino e do Segredo do cliente do principal de serviço. Para localizar estas informações, selecione a sua Microsoft Entra ID.
Em seguida, selecione Registros de aplicativo.
Selecione a sua aplicação e localize as informações necessárias:
Nome
ID de Cliente (ou ID da Aplicação)
ID do Inquilino (ou ID do Diretório)
-
Agora, tem de atribuir as funções relevantes do Microsoft Purview ao seu principal de serviço. Para tal, aceda à sua instância do Microsoft Purview. Selecione Abrir o portal de governação do Microsoft Purview ou abra diretamente o portal de governação do Microsoft Purview e escolha a instância que implementou.
No portal de governação do Microsoft Purview, selecione Mapa de dados e, em seguida, Coleções:
Selecione a coleção com a qual pretende trabalhar e aceda ao separador Atribuições de funções. Adicione o principal de serviço nas seguintes funções:
- Administradores de coleções
- Administradores da origem de dados
- Curadores de dados
- Leitores de dados
Para cada função, selecione o botão Editar atribuições de função e selecione a função à qual pretende adicionar o principal de serviço. Em alternativa, selecione o botão Adicionar junto a cada função e adicione o principal de serviço ao procurar o respetivo nome ou ID de Cliente, conforme mostrado abaixo:
Instalar os pacotes Python
- Abrir uma nova linha de comandos ou terminal
- Instale o pacote de identidade do Azure para autenticação:
pip install azure-identity
- Instale o pacote de Cliente de Análise do Microsoft Purview:
pip install azure-purview-scanning
- Instale o pacote de Cliente de Administração do Microsoft Purview:
pip install azure-purview-administration
- Instale o pacote de Cliente do Microsoft Purview:
pip install azure-purview-catalog
- Instale o pacote de Conta do Microsoft Purview:
pip install azure-purview-account
- Instale o pacote do Azure Core:
pip install azure-core
Criar ficheiro de script python
Crie um ficheiro de texto simples e guarde-o como um script python com o sufixo .py. Por exemplo: tutorial.py.
Instanciar um cliente de Análise, Catálogo e Administração
Nesta secção, irá aprender a instanciar:
- Um cliente de análise útil para registar origens de dados, criar e gerir regras de análise, acionar uma análise, etc.
- Um cliente de catálogo útil para interagir com o catálogo através da pesquisa, navegação nos recursos detetados, identificação da sensibilidade dos seus dados, etc.
- Um cliente de administração é útil para interagir com a própria Mapa de Dados do Microsoft Purview, para operações como a listagem de coleções.
Primeiro, tem de se autenticar no seu Microsoft Entra ID. Para tal, irá utilizar o segredo do cliente que criou.
Comece com as instruções de importação necessárias: os 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
Especifique as seguintes informações no código:
- ID de Cliente (ou ID da Aplicação)
- ID do Inquilino (ou ID do Diretório)
- Segredo do cliente
client_id = "<your client id>" client_secret = "<your client secret>" tenant_id = "<your tenant id>"
Especifique o ponto final:
Importante
O valor do ponto final será diferente consoante o portal do Microsoft Purview que estiver a utilizar. Ponto final para o portal de governação clássico do Microsoft Purview:
https://{your_purview_account_name}.purview.azure.com/
Ponto final para o novo portal do Microsoft Purview:https://api.purview-service.microsoft.com
Analisar o ponto final do portal de governação clássico do Microsoft Purview:
https://{your_purview_account_name}.scan.purview.azure.com/
Ponto final para o novo portal do Microsoft Purview:https://api.scan.purview-service.microsoft.com
purview_endpoint = "<endpoint>" purview_scan_endpoint = "<scan endpoint>"
Agora 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 dos nossos scripts começarão com estes mesmos passos, pois precisaremos destes clientes para interagir com a conta.
Registar uma origem de dados
Nesta secção, irá registar o Armazenamento de Blobs.
Tal como discutimos na secção anterior, primeiro irá importar os clientes de que precisa para aceder à sua conta do Microsoft Purview. Importe também o pacote de resposta de erros do Azure para que possa resolver problemas e o ClientSecretCredential para construir as 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
Reúna o ID de recurso da sua conta de armazenamento ao seguir este guia: obter o ID de recurso de uma conta de armazenamento.
Em seguida, no ficheiro Python, defina as seguintes informações para poder registar o armazenamento de Blobs através de programação:
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>"
Indique o nome da coleção onde pretende registar o armazenamento de blobs. (Deve ser a mesma coleção onde aplicou permissões anteriormente. Se não estiver, aplique primeiro permissões a esta coleção.) Se for a coleção de raiz, utilize o mesmo nome que a sua instância do Microsoft Purview.
collection_name = "<name of your collection>"
Crie uma função para construir as credenciais para aceder à 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
Todas as coleções na Mapa de Dados do Microsoft Purview têm um nome amigável e um nome.
- O nome amigável é aquele que vê na coleção. Por exemplo: Vendas.
- O nome de todas as coleções (exceto a coleção de raiz) é um nome de seis carateres atribuído pelo mapa de dados.
O Python precisa deste nome de seis carateres para referenciar as sub coleções. Para converter automaticamente o seu nome amigável para o nome de coleção de seis carateres necessário no script, adicione este bloco de código:
Importante
O valor do ponto final será diferente consoante o portal do Microsoft Purview que estiver a utilizar. Ponto final do portal de governação clássico do Microsoft Purview: purview.azure.com/ Ponto Final para o novo portal do Microsoft Purview: purview.microsoft.com/
Por isso, se estiver a utilizar o novo portal, o valor do ponto final será semelhante a: "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"]
Para ambos os clientes e, dependendo das operações, também tem de fornecer um corpo de entrada. Para registar uma origem, terá de fornecer um corpo de entrada para o registo da origem 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" } }
Agora, pode ligar para os seus clientes do Microsoft Purview e registar a origem de dados.
Importante
O valor do ponto final será diferente consoante o portal do Microsoft Purview que estiver a utilizar. Ponto final para o portal de governação clássico do Microsoft Purview:
https://{your_purview_account_name}.purview.azure.com/
Ponto final para o novo portal do Microsoft Purview:https://api.purview-service.microsoft.com
Se estiver a utilizar o portal clássico, o valor do ponto final será:
https://{your_purview_account_name}.scan.purview.azure.com
Se estiver a utilizar o novo portal, o valor do ponto final 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 registo for bem-sucedido, pode ver uma resposta corporal melhorada do cliente.
Nas secções seguintes, irá analisar a origem de dados que registou e procurar no catálogo. Cada um destes scripts será estruturado de forma semelhante a este script de registo.
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)
Analisar a origem de dados
A análise de uma origem de dados pode ser feita em dois passos:
- Criar uma definição de análise
- Acionar uma execução de análise
Neste tutorial, irá utilizar as regras de análise predefinidas para contentores de Armazenamento de Blobs. No entanto, também pode criar regras de análise personalizadas programaticamente com o Cliente de Análise do Microsoft Purview.
Agora, vamos analisar a origem de dados que registou acima.
Adicione uma instrução de importação para gerar um identificador exclusivo, chame o cliente de análise do Microsoft Purview, o cliente de administração do Microsoft Purview, o pacote de resposta de erros do Azure para conseguir resolver problemas e a credencial do segredo do cliente para recolher as 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
Crie um cliente de análise com as 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)
Adicione o código para recolher o nome interno da coleção. (Para obter mais informações, consulte a secçã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"]
Em seguida, crie uma definição de análise:
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)
Agora que a análise está definida, pode acionar uma execução de análise com um ID exclusivo:
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)
Catálogo de pesquisa
Quando uma análise estiver concluída, é provável que os recursos tenham sido detetados e até classificados. Este processo pode demorar algum tempo a ser concluído após uma análise, pelo que poderá ter de aguardar antes de executar esta próxima parte do código. Aguarde até que a análise seja apresentada como concluída e os recursos sejam apresentados no Catálogo de Dados do Microsoft Purview.
Assim que os recursos estiverem prontos, pode utilizar o cliente do Catálogo do Microsoft Purview para procurar em todo o catálogo.
Desta vez, tem de importar o cliente do catálogo em vez do cliente de análise. 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
Crie uma função para obter as credenciais para aceder à sua conta do Microsoft Purview e instanciar 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)
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, só especifica palavras-chave, mas tenha em atenção que pode adicionar muitos outros campos para especificar ainda mais a sua consulta.
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)
Eliminar uma origem de dados
Nesta secção, irá aprender a eliminar a origem de dados que registou anteriormente. Esta operação é bastante simples e é feita com o cliente de análise.
Importe o cliente de análise . 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
Crie uma função para obter as credenciais para aceder à sua conta do Microsoft Purview e instanciar o cliente de análise.
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)
Elimine a origem 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)