Esercitazione: Usare Microsoft Purview Python SDK
Questa esercitazione presenta l'uso di Microsoft Purview Python SDK. È possibile usare l'SDK per eseguire tutte le operazioni più comuni di Microsoft Purview a livello di codice, anziché tramite il portale di governance di Microsoft Purview.
In questa esercitazione si apprenderà come usare l'SDK per:
- Concedere i diritti necessari per il funzionamento a livello di codice con Microsoft Purview
- Registrare un contenitore di archiviazione BLOB come origine dati in Microsoft Purview
- Definire ed eseguire un'analisi
- Cercare nel catalogo
- Eliminare un'origine dati
Prerequisiti
Per questa esercitazione, è necessario:
- Python 3.6 o versione successiva
- Sottoscrizione di Azure attiva. Se non ne hai uno, puoi crearne uno gratuitamente.
- Un tenant Microsoft Entra associato alla sottoscrizione.
- Un account di archiviazione di Azure. Se non è già disponibile, è possibile seguire la guida introduttiva per crearne una.
- Un account Microsoft Purview. Se non è già disponibile, è possibile seguire la guida introduttiva per crearne una.
- Entità servizio con un segreto client.
Importante
Per questi script il valore dell'endpoint sarà diverso a seconda del portale di Microsoft Purview in uso. Endpoint per il portale di governance di Microsoft Purview classico: purview.azure.com/ Endpoint per il nuovo portale di Microsoft Purview: purview.microsoft.com/
Pertanto, se si usa il nuovo portale, il valore dell'endpoint sarà simile al seguente: "https://consotopurview.scan.purview.microsoft.com"
Concedere a Microsoft Purview l'accesso all'account di archiviazione
Prima di poter analizzare il contenuto dell'account di archiviazione, è necessario assegnare a Microsoft Purview il ruolo corretto.
Passare all'account di archiviazione tramite il portale di Azure.
Selezionare Controllo di accesso (IAM).
Selezionare il pulsante Aggiungi e selezionare Aggiungi assegnazione di ruolo.
Nella finestra successiva cercare il ruolo Lettore BLOB di archiviazione e selezionarlo:
Passare quindi alla scheda Membri e selezionare Seleziona membri:
A destra viene visualizzato un nuovo riquadro. Cercare e selezionare il nome dell'istanza di Microsoft Purview esistente.
È quindi possibile selezionare Rivedi e assegna.
Microsoft Purview ha ora il diritto di lettura necessario per analizzare l'archiviazione BLOB.
Concedere all'applicazione l'accesso all'account Microsoft Purview
In primo luogo, saranno necessari l'ID client, l'ID tenant e il segreto client dell'entità servizio. Per trovare queste informazioni, selezionare il Microsoft Entra ID.
Selezionare quindi Registrazioni app.
Selezionare l'applicazione e individuare le informazioni necessarie:
Nome
ID client (o ID applicazione)
ID tenant (o ID directory)
-
È ora necessario assegnare i ruoli di Microsoft Purview pertinenti all'entità servizio. A tale scopo, accedere all'istanza di Microsoft Purview. Selezionare Apri il portale di governance di Microsoft Purview o aprire direttamente il portale di governance di Microsoft Purview e scegliere l'istanza distribuita.
All'interno del portale di governance di Microsoft Purview selezionare Mappa dati e quindi Raccolte:
Selezionare la raccolta con cui si vuole lavorare e passare alla scheda Assegnazioni di ruolo . Aggiungere l'entità servizio nei ruoli seguenti:
- Amministratori della raccolta
- Amministratori dell'origine dati
- Curatori dei dati
- Lettori di dati
Per ogni ruolo, selezionare il pulsante Modifica assegnazioni di ruolo e selezionare il ruolo a cui si vuole aggiungere l'entità servizio. In alternativa, selezionare il pulsante Aggiungi accanto a ogni ruolo e aggiungere l'entità servizio cercandone il nome o l'ID client come illustrato di seguito:
Installare i pacchetti Python
- Aprire un nuovo prompt dei comandi o un terminale
- Installare il pacchetto di identità di Azure per l'autenticazione:
pip install azure-identity
- Installare il pacchetto Microsoft Purview Scanning Client:
pip install azure-purview-scanning
- Installare il pacchetto client di amministrazione di Microsoft Purview:
pip install azure-purview-administration
- Installare il pacchetto client Microsoft Purview:
pip install azure-purview-catalog
- Installare il pacchetto dell'account Microsoft Purview:
pip install azure-purview-account
- Installare il pacchetto Azure Core:
pip install azure-core
Creare un file di script Python
Creare un file di testo normale e salvarlo come script Python con il suffisso .py. Ad esempio: tutorial.py.
Creare un'istanza di un client di analisi, catalogo e amministrazione
In questa sezione viene illustrato come creare un'istanza di :
- Un client di analisi utile per registrare le origini dati, creare e gestire le regole di analisi, attivare un'analisi e così via.
- Un client di catalogo utile per interagire con il catalogo tramite la ricerca, l'esplorazione degli asset individuati, l'identificazione della sensibilità dei dati e così via.
- Un client di amministrazione è utile per interagire con il Microsoft Purview Data Map stesso, per operazioni come l'elenco di raccolte.
Prima di tutto è necessario eseguire l'autenticazione al Microsoft Entra ID. A tale scopo, si userà il segreto client creato.
Iniziare con le istruzioni di importazione necessarie: i tre client, l'istruzione credentials e un'istruzione di eccezioni di 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
Specificare le informazioni seguenti nel codice:
- ID client (o ID applicazione)
- ID tenant (o ID directory)
- Segreto client
client_id = "<your client id>" client_secret = "<your client secret>" tenant_id = "<your tenant id>"
Specificare l'endpoint:
Importante
Il valore dell'endpoint sarà diverso a seconda del portale di Microsoft Purview in uso. Endpoint per il portale di governance di Microsoft Purview classico:
https://{your_purview_account_name}.purview.azure.com/
Endpoint per il nuovo portale Microsoft Purview:https://api.purview-service.microsoft.com
Eseguire l'analisi dell'endpoint per il portale di governance di Microsoft Purview classico:
https://{your_purview_account_name}.scan.purview.azure.com/
Endpoint per il nuovo portale Di Microsoft Purview:https://api.scan.purview-service.microsoft.com
purview_endpoint = "<endpoint>" purview_scan_endpoint = "<scan endpoint>"
È ora possibile creare un'istanza dei tre client:
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
Molti degli script inizieranno con questi stessi passaggi, perché saranno necessari questi client per interagire con l'account.
Registrare un'origine dati
In questa sezione si registrerà l'archiviazione BLOB.
Come illustrato nella sezione precedente, prima si importeranno i client necessari per accedere all'account Microsoft Purview. Importare anche il pacchetto di risposta agli errori di Azure in modo da poter risolvere i problemi e ClientSecretCredential per costruire le credenziali di 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
Raccogliere l'ID risorsa per l'account di archiviazione seguendo questa guida: ottenere l'ID risorsa per un account di archiviazione.
Nel file Python definire quindi le informazioni seguenti per poter registrare l'archiviazione BLOB a livello di codice:
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>"
Specificare il nome della raccolta in cui si vuole registrare l'archiviazione BLOB. Deve essere la stessa raccolta in cui sono state applicate le autorizzazioni in precedenza. In caso contrario, applicare prima di tutto le autorizzazioni a questa raccolta. Se si tratta della raccolta radice, usare lo stesso nome dell'istanza di Microsoft Purview.
collection_name = "<name of your collection>"
Creare una funzione per costruire le credenziali per accedere all'account 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
Tutte le raccolte nel Microsoft Purview Data Map hanno un nome descrittivo e un nome.
- Il nome descrittivo è quello visualizzato nella raccolta. Ad esempio: Vendite.
- Il nome di tutte le raccolte (ad eccezione della raccolta radice) è un nome di sei caratteri assegnato dalla mappa dati.
Python ha bisogno di questo nome di sei caratteri per fare riferimento a qualsiasi sotto-raccolta. Per convertire automaticamente il nome descrittivo nel nome della raccolta di sei caratteri necessario nello script, aggiungere questo blocco di codice:
Importante
Il valore dell'endpoint sarà diverso a seconda del portale di Microsoft Purview in uso. Endpoint per il portale di governance di Microsoft Purview classico: purview.azure.com/ Endpoint per il nuovo portale di Microsoft Purview: purview.microsoft.com/
Pertanto, se si usa il nuovo portale, il valore dell'endpoint sarà simile al seguente: "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"]
Per entrambi i client e, a seconda delle operazioni, è anche necessario fornire un corpo di input. Per registrare un'origine, è necessario fornire un corpo di input per la registrazione dell'origine dati:
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" } }
È ora possibile chiamare i client Microsoft Purview e registrare l'origine dati.
Importante
Il valore dell'endpoint sarà diverso a seconda del portale di Microsoft Purview in uso. Endpoint per il portale di governance di Microsoft Purview classico:
https://{your_purview_account_name}.purview.azure.com/
Endpoint per il nuovo portale Microsoft Purview:https://api.purview-service.microsoft.com
Se si usa il portale classico, il valore dell'endpoint sarà:
https://{your_purview_account_name}.scan.purview.azure.com
Se si usa il nuovo portale, il valore dell'endpoint sarà: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 il processo di registrazione ha esito positivo, è possibile visualizzare una risposta del corpo arricchita dal client.
Nelle sezioni seguenti si analizzerà l'origine dati registrata e si eseguirà la ricerca nel catalogo. Ognuno di questi script sarà strutturato in modo analogo a questo script di registrazione.
Codice 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)
Analizzare l'origine dati
L'analisi di un'origine dati può essere eseguita in due passaggi:
- Creare una definizione di analisi
- Attivare un'esecuzione di analisi
In questa esercitazione verranno usate le regole di analisi predefinite per i contenitori di archiviazione BLOB. Tuttavia, è anche possibile creare regole di analisi personalizzate a livello di codice con Microsoft Purview Scanning Client.
A questo punto si analizzerà l'origine dati registrata in precedenza.
Aggiungere un'istruzione import per generare l'identificatore univoco, chiamare il client di analisi di Microsoft Purview, il client di amministrazione di Microsoft Purview, il pacchetto di risposta agli errori di Azure per la risoluzione dei problemi e le credenziali del segreto client per raccogliere le credenziali di 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
Creare un client di analisi usando le credenziali:
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)
Aggiungere il codice per raccogliere il nome interno della raccolta. Per altre informazioni, vedere la sezione precedente:
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"]
Creare quindi una definizione di analisi:
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)
Ora che l'analisi è definita, è possibile attivare un'esecuzione dell'analisi con un ID univoco:
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)
Codice 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)
Catalogo di ricerca
Una volta completata un'analisi, è probabile che gli asset siano stati individuati e persino classificati. Il completamento di questo processo può richiedere del tempo dopo un'analisi, quindi potrebbe essere necessario attendere prima di eseguire questa parte successiva del codice. Attendere che l'analisi venga visualizzata completata e che gli asset vengano visualizzati nel Microsoft Purview Data Catalog.
Quando gli asset sono pronti, è possibile usare il client del catalogo Microsoft Purview per eseguire ricerche nell'intero catalogo.
Questa volta è necessario importare il client del catalogo anziché quello di analisi. Includere anche l'errore HTTPResponse e ClientSecretCredential.
from azure.purview.catalog import PurviewCatalogClient from azure.identity import ClientSecretCredential from azure.core.exceptions import HttpResponseError
Creare una funzione per ottenere le credenziali per accedere all'account Microsoft Purview e creare un'istanza del client del catalogo.
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)
Configurare i criteri di ricerca e le parole chiave nel corpo di input:
keywords = "keywords you want to search" body_input={ "keywords": keywords }
Qui si specificano solo parole chiave, ma tenere presente che è possibile aggiungere molti altri campi per specificare ulteriormente la query.
Eseguire una ricerca nel catalogo:
try: response = client_catalog.discovery.query(search_request=body_input) print(response) except HttpResponseError as e: print(e)
Codice 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)
Eliminare un'origine dati
In questa sezione si apprenderà come eliminare l'origine dati registrata in precedenza. Questa operazione è abbastanza semplice e viene eseguita con il client di analisi.
Importare il client di analisi . Includere anche l'errore HTTPResponse e ClientSecretCredential.
from azure.purview.scanning import PurviewScanningClient from azure.identity import ClientSecretCredential from azure.core.exceptions import HttpResponseError
Creare una funzione per ottenere le credenziali per accedere all'account Microsoft Purview e creare un'istanza del client di analisi.
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)
Eliminare l'origine dati:
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)
Codice 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)