Tutorial: Uso del SDK de Python de Microsoft Purview

Este tutorial le presentará el uso del SDK de Python de Microsoft Purview. Puede usar el SDK para realizar todas las operaciones más comunes de Microsoft Purview mediante programación, en lugar de hacerlo a través del portal de gobierno de Microsoft Purview.

En este tutorial, aprenderá cómo usar el SDK para:

Requisitos previos

Para este tutorial, necesitarás:

• Python 3.6 o superior
• Una suscripción activa a Azure. Si no tienes una, puedes crear una gratis.
• Un inquilino de Microsoft Entra asociado a su suscripción.
• Una cuenta de almacenamiento de Azure. Si aún no tienes una, puedes seguir nuestra guía de inicio rápido para crear una.
• Una cuenta de Microsoft Purview. Si aún no tienes una, puedes seguir nuestra guía de inicio rápido para crear una.
• Una entidad de servicio con un secreto de cliente.

Conceder acceso Microsoft Purview a la cuenta de almacenamiento

Antes de poder examinar el contenido de la cuenta de almacenamiento, debe asignar a Microsoft Purview el rol correcto.

1. Vaya a su cuenta de almacenamiento mediante Azure Portal.

2. Seleccione Access Control (IAM).

3. Haga clic en el botón Agregar y seleccione Agregar asignación de roles.

   

4. En la ventana siguiente, busque el rol de lector de blobs de Storage y selecciónelo:

   

5. A continuación, vaya a la pestaña Miembros y seleccione Seleccionar miembros:

   

6. Aparecerá un nuevo panel a la derecha. Busca y selecciona el nombre de la instancia de Microsoft Purview existente.

7. A continuación, puede seleccionar Revisar + Asignar.

Microsoft Purview ahora tiene el derecho de lectura necesario para examinar el almacenamiento de blobs.

Conceda a la aplicación acceso a su cuenta de Microsoft Purview

1. En primer lugar, necesitará el identificador de cliente, el identificador de inquilino y el secreto de cliente de la entidad de servicio. Para encontrar esta información, seleccione su Microsoft Entra ID.

2. A continuación, seleccione Registros de aplicaciones.

3. Seleccione la aplicación y busque la información requerida:

   • Nombre

   • Id. de cliente (o id. de aplicación)

   • Id. de inquilino (o Id. de directorio)

     

   • Secreto de cliente

     

4. Ahora debe asignar los roles de Microsoft Purview pertinentes a la entidad de servicio. Para ello, accede a tu cuenta de Microsoft Purview. Seleccione Abrir el portal de gobierno de Microsoft Purview o abrir el portal de gobierno de Microsoft Purview directamente y elija la instancia que implementó.

5. Dentro del portal de gobierno de Microsoft Purview, selecciona Asignación de datos y, a continuación, Colecciones:

   

6. Seleccione la colección con la que desea trabajar y vaya a la pestaña Asignaciones de roles . Agregue la entidad de servicio en los siguientes roles:

   • Administradores de colecciones
   • Administradores del origen de datos
   • Conservadores de datos
   • Lectores de datos

7. Para cada rol, seleccione el botón Editar asignaciones de roles y seleccione el rol al que desea agregar la entidad de servicio. O bien, seleccione el botón Agregar junto a cada rol y agregue la entidad de servicio buscando su nombre o id. de cliente como se muestra a continuación:

   

Instalar los paquetes de Python

1. Abra un nuevo símbolo del sistema o terminal
2. Instale el paquete de identidad de Azure para la autenticación:

   pip install azure-identity
   

3. Instale el paquete de cliente de examen de Microsoft Purview:

   pip install azure-purview-scanning
   

4. Instale el paquete de cliente de administración de Microsoft Purview:

   pip install azure-purview-administration
   

5. Instala el paquete de cliente de Microsoft Purview:

   pip install azure-purview-catalog
   

6. Instala el paquete de cuenta de Microsoft Purview:

   pip install azure-purview-account
   

7. Instale el paquete de Azure Core:

   pip install azure-core
   

Crear archivo de script de Python

Cree un archivo de texto sin formato y guárdelo como una secuencia de comandos de Python con el sufijo .py. Por ejemplo: tutorial.py.

Crear una instancia de un cliente de análisis, catálogo y administración

En esta sección, aprenderá a crear instancias:

• Un cliente de escaneo útil para registrar fuentes de datos, crear y administrar reglas de escaneo, activar un escaneo, etc.
• Un cliente de catálogo útil para interactuar con el catálogo a través de la búsqueda, la navegación por los activos descubiertos, la identificación de la confidencialidad de sus datos, etc.
• Un cliente de administración es útil para interactuar con el propio Mapa de datos de Microsoft Purview, para operaciones como enumerar colecciones.

En primer lugar, debe autenticarse en su Microsoft Entra ID. Para ello, deberá usar el secreto de cliente que ha creado.

1. Comience con las instrucciones import necesarias: nuestros tres clientes, la instrucción de credenciales y una instrucción de excepciones de 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 la siguiente información en el código:

   • Id. de cliente (o id. de aplicación)
   • Id. de inquilino (o Id. de directorio)
   • Secreto de cliente

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

3. Especifique el punto de conexión:

   Importante

   El valor del punto de conexión será diferente en función del portal de Microsoft Purview que se use. Punto de conexión para el portal de gobierno clásico de Microsoft Purview: https://{your_purview_account_name}.purview.azure.com/ Punto de conexión para el nuevo portal de Microsoft Purview: https://api.purview-service.microsoft.com

   Examinar el punto de conexión para el portal de gobierno clásico de Microsoft Purview: https://{your_purview_account_name}.scan.purview.azure.com/ Punto de conexión para el nuevo portal de Microsoft Purview: https://api.scan.purview-service.microsoft.com

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

4. Ahora puede instanciar los tres 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
   

Muchos de nuestros scripts comenzarán con estos mismos pasos, ya que necesitaremos que estos clientes interactúen con la cuenta.

Registrar un origen de datos

En esta sección, registrará su almacenamiento de blobs.

1. Como explicamos en la sección anterior, primero importarás los clientes que necesitarás para acceder a tu cuenta de Microsoft Purview. Importe también el paquete de respuesta a errores de Azure para que pueda solucionar problemas y ClientSecretCredential para construir las credenciales de 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. Recopile el identificador de recurso de la cuenta de almacenamiento siguiendo esta guía: obtener el identificador de recurso de una cuenta de almacenamiento.

3. A continuación, en el archivo de Python, defina la siguiente información para poder registrar el almacenamiento de blobs mediante programación:

   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. Proporcione el nombre de la colección en la que desea registrar el almacenamiento de blobs. (Debe ser la misma colección en la que aplicó permisos anteriormente. Si no es así, aplique primero permisos a esta colección). Si es la colección raíz, usa el mismo nombre que tu instancia de Microsoft Purview.

   collection_name = "<name of your collection>"
   

5. Cree una función para crear las credenciales para acceder a su cuenta de 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 las colecciones del Mapa de datos de Microsoft Purview tienen un nombre descriptivo y un nombre.

   • El nombre descriptivo es el que se ve en la colección. Por ejemplo: Ventas.
   • El nombre de todas las colecciones (excepto la colección raíz) es un nombre de seis caracteres asignado por el mapa de datos.

   Python necesita este nombre de seis caracteres para hacer referencia a cualquier subcolección. Para convertir automáticamente el nombre descriptivo al nombre de colección de seis caracteres necesario en el script, agregue este bloque de código:

   Importante

   El valor del punto de conexión será diferente en función del portal de Microsoft Purview que se use. Punto de conexión para el portal de gobierno clásico de Microsoft Purview: purview.azure.com/ Punto de conexión para el nuevo portal de Microsoft Purview: purview.microsoft.com/

   Por lo tanto, si usa el nuevo portal, el valor del punto de conexión 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 clientes y en función de las operaciones, también debe proporcionar un cuerpo de entrada. Para registrar un origen, deberá proporcionar un cuerpo de entrada para el registro del origen de datos:

   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. Ahora puede llamar a los clientes de Microsoft Purview y registrar el origen de datos.

   Importante

   El valor del punto de conexión será diferente en función del portal de Microsoft Purview que se use. Punto de conexión para el portal de gobierno clásico de Microsoft Purview: https://{your_purview_account_name}.purview.azure.com/ Punto de conexión para el nuevo portal de Microsoft Purview: https://api.purview-service.microsoft.com

   Si usa el portal clásico, el valor de punto de conexión será: https://{your_purview_account_name}.scan.purview.azure.com Si usa el nuevo portal, el valor de punto de conexión 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)
   

Cuando el proceso de registro se realiza correctamente, puede ver una respuesta de cuerpo enriquecido del cliente.

En las secciones siguientes, examinará el origen de datos que registró y buscará en el catálogo. Cada uno de estos scripts se estructurará de forma similar 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)

Examinar el origen de datos

El examen de un origen de datos se puede realizar en dos pasos:

1. Crear una definición de examen
2. Desencadenar una ejecución de examen

En este tutorial, usará las reglas de análisis predeterminadas para los contenedores de Blob Storage. Sin embargo, también puedes crear reglas de examen personalizadas mediante programación con el cliente de examen de Microsoft Purview.

Ahora echemos un vistazo al origen de datos que registró anteriormente.

1. Agregue una instrucción import para generar un identificador único, llame al cliente de examen de Microsoft Purview, al cliente de administración de Microsoft Purview, al paquete de respuesta a errores de Azure para poder solucionar los problemas y a la credencial secreta de cliente para recopilar sus credenciales de 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. Cree un cliente de análisis con sus credenciales:

   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. Agregue el código para recopilar el nombre interno de la colección. (Para obtener más información, consulte la sección 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. A continuación, cree una definición de examen:

   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. Ahora que el examen está definido, puede desencadenar una ejecución de examen con un identificador único:

   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)

Buscar en el catálogo

Una vez que se completa un escaneo, es probable que se hayan descubierto e incluso clasificado activos. Este proceso puede tardar algún tiempo en completarse después de un examen, por lo que es posible que deba esperar antes de ejecutar la siguiente parte del código. Espera a que el examen se muestre completado y a que los activos aparezcan en el Catálogo unificado de Microsoft Purview.

Una vez que los recursos estén listos, puede usar el cliente de catálogo de Microsoft Purview para buscar en todo el catálogo.

1. Esta vez debe importar el cliente de catálogo en lugar del de escaneo. Incluya también el error HTTPResponse y ClientSecretCredential.

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

2. Cree una función para obtener las credenciales para acceder a su cuenta de Microsoft Purview y crear una instancia del cliente de 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. Configura los criterios de búsqueda y las palabras clave en el cuerpo de entrada:

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

   Aquí solo se especifican palabras clave, pero ten en cuenta que puedes agregar muchos otros campos para especificar aún más tu consulta.

4. Buscar en el 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 un origen de datos

En esta sección, aprenderá a eliminar el origen de datos que registró anteriormente. Esta operación es bastante simple y se realiza con el cliente de escaneo.

1. Importe el cliente de análisis . Incluya también el error HTTPResponse y ClientSecretCredential.

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

2. Cree una función para obtener las credenciales para acceder a su cuenta de Microsoft Purview y crear una instancia del cliente de análisis.

   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. Elimine el origen de datos:

       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)