Tutorial: Uso del SDK de Python de Microsoft Purview
En este tutorial se 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 a través del portal de gobernanza de Microsoft Purview.
En este tutorial, aprenderá a usar el SDK para:
- Concesión de los derechos necesarios para trabajar mediante programación con Microsoft Purview
- Registro de un contenedor de Blob Storage como origen de datos en Microsoft Purview
- Definición y ejecución de un examen
- Buscar en el catálogo
- Eliminación de un origen de datos
Requisitos previos
Para este tutorial, necesitará lo siguiente:
- Python 3.6 o superior
- Una suscripción de Azure activa. Si no tiene uno, puede crear uno de forma gratuita.
- Un inquilino de Microsoft Entra asociado a la suscripción.
- Una cuenta de Azure Storage. Si aún no tiene uno, puede seguir nuestra guía de inicio rápido para crear una.
- Una cuenta de Microsoft Purview. Si aún no tiene uno, puede seguir nuestra guía de inicio rápido para crear una.
- Una entidad de servicio con un secreto de cliente.
Importante
Para estos scripts, el valor del punto de conexión será diferente en función del portal de Microsoft Purview que use. Punto de conexión para el portal de gobernanza de Microsoft Purview clásico: punto de conexión purview.azure.com/ 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á similar al siguiente: "https://consotopurview.scan.purview.microsoft.com"
Conceder a Microsoft Purview acceso a la cuenta de almacenamiento
Antes de poder examinar el contenido de la cuenta de almacenamiento, debe asignar a Microsoft Purview el rol correcto.
Vaya a la cuenta de almacenamiento a través de la Azure Portal.
Seleccione Access Control (IAM).
Seleccione el botón Agregar y seleccione Agregar asignación de roles.
En la siguiente ventana, busque el rol Lector de blobs de Storage y selecciónelo:
A continuación, vaya a la pestaña Miembros y seleccione Seleccionar miembros:
Aparece un nuevo panel a la derecha. Busque y seleccione el nombre de la instancia de Microsoft Purview existente.
Después, puede seleccionar Revisar y asignar.
Microsoft Purview ahora tiene el derecho de lectura necesario para examinar Blob Storage.
Conceder a la aplicación acceso a su cuenta de Microsoft Purview
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 el Microsoft Entra ID.
A continuación, seleccione Registros de aplicaciones.
Seleccione la aplicación y busque la información necesaria:
Nombre
Id. de cliente (o id. de aplicación)
Identificador de inquilino (o id. de directorio)
-
Ahora debe proporcionar los roles de Microsoft Purview pertinentes a la entidad de servicio. Para ello, acceda a la instancia de Microsoft Purview. Seleccione Abrir el portal de gobernanza de Microsoft Purview o abra directamente el portal de gobernanza de Microsoft Purview y elija la instancia que implementó.
En el portal de gobernanza de Microsoft Purview, seleccione Mapa de datos y, a continuación, Colecciones:
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 recopilación
- Administradores del origen de datos
- Conservadores de datos
- Lectores de datos
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 situado junto a cada rol y agregue la entidad de servicio buscando su nombre o identificador de cliente como se muestra a continuación:
Instalación de los paquetes de Python
- Abrir un nuevo símbolo del sistema o terminal
- Instale el paquete de identidad de Azure para la autenticación:
pip install azure-identity
- Instale el paquete cliente de examen de Microsoft Purview:
pip install azure-purview-scanning
- Instale el paquete cliente de administración de Microsoft Purview:
pip install azure-purview-administration
- Instale el paquete cliente de Microsoft Purview:
pip install azure-purview-catalog
- Instale el paquete de la cuenta de Microsoft Purview:
pip install azure-purview-account
- Instale el paquete de Azure Core:
pip install azure-core
Creación de un archivo de script de Python
Cree un archivo de texto sin formato y guárdelo como un script de Python con el sufijo .py. Por ejemplo: tutorial.py.
Crear una instancia de un cliente de examen, catálogo y administración
En esta sección, aprenderá a crear instancias de:
- Un cliente de examen útil para registrar orígenes de datos, crear y administrar reglas de examen, desencadenar un examen, etc.
- Un cliente de catálogo útil para interactuar con el catálogo mediante la búsqueda, la exploración de los recursos detectados, la identificación de la confidencialidad de los 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 el Microsoft Entra ID. Para ello, usará el secreto de cliente que creó.
Comience con las instrucciones de importación necesarias: nuestros tres clientes, la instrucción credentials 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
Especifique la siguiente información en el código:
- Id. de cliente (o id. de aplicación)
- Identificador de inquilino (o id. de directorio)
- Secreto de cliente
client_id = "<your client id>" client_secret = "<your client secret>" tenant_id = "<your tenant id>"
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 esté usando. Punto de conexión para el portal de gobernanza de Microsoft Purview clásico:
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 gobernanza de Microsoft Purview clásico:
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>"
Ahora puede crear instancias de 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.
Registro de un origen de datos
En esta sección, registrará el almacenamiento de blobs.
Como se ha explicado en la sección anterior, primero importará los clientes que necesitará para acceder a su cuenta de Microsoft Purview. Importe también el paquete de respuesta de error 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
Recopile el identificador de recurso de la cuenta de almacenamiento siguiendo esta guía: obtenga el identificador de recurso de una cuenta de almacenamiento.
A continuación, en el archivo de Python, defina la siguiente información para poder registrar Blob Storage 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>"
Proporcione el nombre de la colección donde desea registrar el almacenamiento de blobs. (Debe ser la misma colección donde aplicó permisos anteriormente. Si no es así, aplique primero permisos a esta colección). Si es la colección raíz, use el mismo nombre que la instancia de Microsoft Purview.
collection_name = "<name of your collection>"
Cree una función para construir las credenciales para acceder a la 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
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 las subcolecciones. Para convertir automáticamente el nombre descriptivo en el nombre de la 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 esté usando. Punto de conexión para el portal de gobernanza de Microsoft Purview clásico: punto de conexión purview.azure.com/ 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á similar al siguiente: "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 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" } }
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 esté usando. Punto de conexión para el portal de gobernanza de Microsoft Purview clásico:
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 del punto de conexión será:
https://{your_purview_account_name}.scan.purview.azure.com
Si usa el nuevo portal, el valor del 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 del 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:
- Creación de una definición de examen
- Desencadenamiento de una ejecución de examen
En este tutorial, usará las reglas de examen predeterminadas para contenedores de Blob Storage. Sin embargo, también puede crear reglas de examen personalizadas mediante programación con el cliente de examen de Microsoft Purview.
Ahora vamos a examinar el origen de datos que registró anteriormente.
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 de errores de Azure para poder solucionar problemas y a la credencial de secreto de cliente para recopilar las 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
Cree un cliente de examen 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)
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"]
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)
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)
Catálogo de búsqueda
Una vez completado un examen, es probable que los recursos se hayan detectado e incluso clasificado. Este proceso puede tardar algún tiempo en completarse después de un examen, por lo que es posible que tenga que esperar antes de ejecutar esta siguiente parte del código. Espere a que el examen se muestre completado y que los recursos aparezcan en el Catálogo de datos de Microsoft Purview.
Una vez que los recursos estén listos, puede usar el cliente del catálogo de Microsoft Purview para buscar en todo el catálogo.
Esta vez debe importar el cliente de catálogo en lugar del de examen. 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
Cree una función para obtener las credenciales para acceder a la 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)
Configure 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 tenga en cuenta que puede agregar muchos otros campos para especificar aún más la consulta.
Busque 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)
Eliminación de un origen de datos
En esta sección, aprenderá a eliminar el origen de datos que registró anteriormente. Esta operación es bastante sencilla y se realiza con el cliente de examen.
Importe el cliente de examen . 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
Cree una función para obtener las credenciales para acceder a la cuenta de Microsoft Purview y crear una instancia del cliente de examen.
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 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)