Tutoriel : Utiliser le Kit de développement logiciel (SDK) Python Microsoft Purview
Ce tutoriel vous présente l’utilisation du Kit de développement logiciel (SDK) Python Microsoft Purview. Vous pouvez utiliser le Kit de développement logiciel (SDK) pour effectuer toutes les opérations Microsoft Purview les plus courantes par programmation, plutôt que via le portail de gouvernance Microsoft Purview.
Dans ce tutoriel, vous allez découvrir comment nous le KIT de développement logiciel (SDK) :
- Accorder les droits requis pour travailler par programmation avec Microsoft Purview
- Inscrire un conteneur Stockage Blob en tant que source de données dans Microsoft Purview
- Définir et exécuter une analyse
- Rechercher dans le catalogue
- Supprimer une source de données
Configuration requise
Pour ce didacticiel, vous avez besoin des éléments suivants :
- Python 3.6 ou version ultérieure
- Un abonnement Azure actif. Si vous n’en avez pas, vous pouvez en créer un gratuitement.
- Un locataire Microsoft Entra associé à votre abonnement.
- Un compte de stockage Azure. Si vous n’en avez pas encore, vous pouvez suivre notre guide de démarrage rapide pour en créer un.
- Un compte Microsoft Purview. Si vous n’en avez pas encore, vous pouvez suivre notre guide de démarrage rapide pour en créer un.
- Un principal de service avec une clé secrète client.
Importante
Pour ces scripts, votre valeur de point de terminaison sera différente en fonction du portail Microsoft Purview que vous utilisez. Point de terminaison pour le portail de gouvernance Microsoft Purview classique : point de terminaison purview.azure.com/ pour le nouveau portail Microsoft Purview : purview.microsoft.com/
Par conséquent, si vous utilisez le nouveau portail, la valeur de votre point de terminaison sera quelque chose comme : «https://consotopurview.scan.purview.microsoft.com" ;
Accorder à Microsoft Purview l’accès au compte de stockage
Avant de pouvoir analyser le contenu du compte de stockage, vous devez attribuer le rôle approprié à Microsoft Purview.
Accédez à votre compte de stockage via le Portail Azure.
Sélectionnez Access Control (IAM).
Sélectionnez le bouton Ajouter, puis Ajouter une attribution de rôle.
Dans la fenêtre suivante, recherchez le rôle Lecteur d’objets blob de stockage et sélectionnez-le :
Ensuite, accédez à l’onglet Membres et sélectionnez Sélectionner les membres :
Un nouveau volet s’affiche à droite. Recherchez et sélectionnez le nom de votre instance Microsoft Purview existant.
Vous pouvez ensuite sélectionner Vérifier + affecter.
Microsoft Purview dispose désormais du droit de lecture requis pour analyser votre stockage Blob.
Accorder à votre application l’accès à votre compte Microsoft Purview
Tout d’abord, vous aurez besoin de l’ID client, de l’ID de locataire et de la clé secrète client de votre principal de service. Pour trouver ces informations, sélectionnez votre Microsoft Entra ID.
Sélectionnez ensuite inscriptions d'applications.
Sélectionnez votre application et recherchez les informations requises :
Nom
ID client (ou ID d’application)
ID de locataire (ou ID d’annuaire)
-
Vous devez maintenant attribuer les rôles Microsoft Purview appropriés à votre principal de service. Pour ce faire, accédez à votre instance Microsoft Purview. Sélectionnez Ouvrir le portail de gouvernance Microsoft Purview ou ouvrez directement le portail de gouvernance de Microsoft Purview et choisissez le instance que vous avez déployé.
Dans le portail de gouvernance Microsoft Purview, sélectionnez Mappage de données, puis Collections :
Sélectionnez la collection que vous souhaitez utiliser, puis accédez à l’onglet Attributions de rôles . Ajoutez le principal de service dans les rôles suivants :
- Administrateurs de collections
- Administrateurs de source de données
- Conservateurs de données
- Lecteurs de données
Pour chaque rôle, sélectionnez le bouton Modifier les attributions de rôles , puis sélectionnez le rôle auquel vous souhaitez ajouter le principal de service. Vous pouvez également sélectionner le bouton Ajouter en regard de chaque rôle, puis ajouter le principal de service en recherchant son nom ou son ID client, comme indiqué ci-dessous :
Installer les packages Python
- Ouvrir une nouvelle invite de commandes ou un nouveau terminal
- Installez le package d’identité Azure pour l’authentification :
pip install azure-identity
- Installez le package client d’analyse Microsoft Purview :
pip install azure-purview-scanning
- Installez le package du client d’administration Microsoft Purview :
pip install azure-purview-administration
- Installez le package du client Microsoft Purview :
pip install azure-purview-catalog
- Installez le package de compte Microsoft Purview :
pip install azure-purview-account
- Installez le package Azure Core :
pip install azure-core
Créer un fichier de script Python
Créez un fichier texte brut et enregistrez-le en tant que script Python avec le suffixe .py. Par exemple : tutorial.py.
Instancier un client d’analyse, de catalogue et d’administration
Dans cette section, vous allez apprendre à instancier :
- Client d’analyse utile pour inscrire des sources de données, créer et gérer des règles d’analyse, déclencher une analyse, etc.
- Un client de catalogue utile pour interagir avec le catalogue via la recherche, la navigation dans les ressources découvertes, l’identification de la sensibilité de vos données, etc.
- Un client d’administration est utile pour interagir avec le Mappage de données Microsoft Purview lui-même, pour des opérations telles que la liste des collections.
Vous devez d’abord vous authentifier auprès de votre Microsoft Entra ID. Pour cela, vous allez utiliser la clé secrète client que vous avez créée.
Commencez par les instructions d’importation requises : nos trois clients, l’instruction d’informations d’identification et une instruction d’exceptions 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
Spécifiez les informations suivantes dans le code :
- ID client (ou ID d’application)
- ID de locataire (ou ID d’annuaire)
- Clé secrète client
client_id = "<your client id>" client_secret = "<your client secret>" tenant_id = "<your tenant id>"
Spécifiez votre point de terminaison :
Importante
La valeur de votre point de terminaison sera différente en fonction du portail Microsoft Purview que vous utilisez. Point de terminaison pour le portail de gouvernance Microsoft Purview classique :
https://{your_purview_account_name}.purview.azure.com/
Point de terminaison du nouveau portail Microsoft Purview :https://api.purview-service.microsoft.com
Analyser le point de terminaison pour le portail de gouvernance Microsoft Purview classique :
https://{your_purview_account_name}.scan.purview.azure.com/
Point de terminaison du nouveau portail Microsoft Purview :https://api.scan.purview-service.microsoft.com
purview_endpoint = "<endpoint>" purview_scan_endpoint = "<scan endpoint>"
Vous pouvez maintenant instancier les trois clients :
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
La plupart de nos scripts commencent par ces mêmes étapes, car nous aurons besoin que ces clients interagissent avec le compte.
Inscrire une source de données
Dans cette section, vous allez inscrire votre Stockage Blob.
Comme nous l’avons vu dans la section précédente, vous allez d’abord importer les clients dont vous aurez besoin pour accéder à votre compte Microsoft Purview. Importez également le package de réponse d’erreur Azure pour pouvoir résoudre les problèmes, et clientSecretCredential pour construire vos informations d’identification 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
Collectez l’ID de ressource pour votre compte de stockage en suivant ce guide : obtenir l’ID de ressource d’un compte de stockage.
Ensuite, dans votre fichier Python, définissez les informations suivantes pour pouvoir inscrire le stockage Blob par programme :
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>"
Indiquez le nom de la collection dans laquelle vous souhaitez inscrire votre stockage d’objets blob. (Il doit s’agir de la même collection que celle où vous avez appliqué des autorisations précédemment. Si ce n’est pas le cas, appliquez d’abord les autorisations à cette collection.) S’il s’agit de la collection racine, utilisez le même nom que votre instance Microsoft Purview.
collection_name = "<name of your collection>"
Créez une fonction pour construire les informations d’identification pour accéder à votre compte 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
Toutes les collections du Mappage de données Microsoft Purview ont un nom convivial et un nom.
- Le nom convivial est celui que vous voyez sur la collection. Par exemple : Ventes.
- Le nom de toutes les collections (à l’exception de la collection racine) est un nom de six caractères attribué par le mappage de données.
Python a besoin de ce nom de six caractères pour référencer toutes les sous-collections. Pour convertir automatiquement votre nom convivial en nom de collection de six caractères nécessaire dans votre script, ajoutez ce bloc de code :
Importante
La valeur de votre point de terminaison sera différente en fonction du portail Microsoft Purview que vous utilisez. Point de terminaison pour le portail de gouvernance Microsoft Purview classique : point de terminaison purview.azure.com/ pour le nouveau portail Microsoft Purview : purview.microsoft.com/
Par conséquent, si vous utilisez le nouveau portail, la valeur de votre point de terminaison sera quelque chose comme : «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"]
Pour les deux clients, et en fonction des opérations, vous devez également fournir un corps d’entrée. Pour inscrire une source, vous devez fournir un corps d’entrée pour l’inscription de la source de données :
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" } }
Vous pouvez maintenant appeler vos clients Microsoft Purview et inscrire la source de données.
Importante
La valeur de votre point de terminaison sera différente en fonction du portail Microsoft Purview que vous utilisez. Point de terminaison pour le portail de gouvernance Microsoft Purview classique :
https://{your_purview_account_name}.purview.azure.com/
Point de terminaison du nouveau portail Microsoft Purview :https://api.purview-service.microsoft.com
Si vous utilisez le portail classic, votre valeur de point de terminaison sera :
https://{your_purview_account_name}.scan.purview.azure.com
Si vous utilisez le nouveau portail, votre valeur de point de terminaison sera :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)
Lorsque le processus d’inscription réussit, vous pouvez voir une réponse de corps enrichie du client.
Dans les sections suivantes, vous allez analyser la source de données que vous avez inscrite et effectuer une recherche dans le catalogue. Chacun de ces scripts sera structuré de la même façon que ce script d’inscription.
Code complet
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)
Analyser la source de données
L’analyse d’une source de données peut être effectuée en deux étapes :
- Créer une définition d’analyse
- Déclencher une exécution d’analyse
Dans ce tutoriel, vous allez utiliser les règles d’analyse par défaut pour les conteneurs de stockage Blob. Toutefois, vous pouvez également créer des règles d’analyse personnalisées par programmation avec le client d’analyse Microsoft Purview.
À présent, nous allons analyser la source de données que vous avez inscrite ci-dessus.
Ajoutez une instruction d’importation pour générer un identificateur unique, appelez le client d’analyse Microsoft Purview, le client d’administration Microsoft Purview, le package de réponse d’erreur Azure pour pouvoir résoudre les problèmes et les informations d’identification de la clé secrète client pour collecter vos informations d’identification 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
Créez un client d’analyse à l’aide de vos informations d’identification :
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)
Ajoutez le code pour collecter le nom interne de votre collection. (Pour plus d’informations, consultez la section précédente) :
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"]
Ensuite, créez une définition d’analyse :
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)
Maintenant que l’analyse est définie, vous pouvez déclencher une exécution d’analyse avec un ID unique :
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)
Code complet
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)
Catalogue de recherche
Une fois l’analyse terminée, il est probable que des ressources aient été découvertes et même classifiées. Ce processus peut prendre un certain temps après une analyse. Vous devrez donc peut-être attendre avant d’exécuter cette partie de code suivante. Attendez que votre analyse s’affiche terminée et que les ressources apparaissent dans le Catalogue de données Microsoft Purview.
Une fois les ressources prêtes, vous pouvez utiliser le client catalogue Microsoft Purview pour effectuer une recherche dans l’ensemble du catalogue.
Cette fois, vous devez importer le client de catalogue au lieu de l’analyse. Incluez également l’erreur HTTPResponse et ClientSecretCredential.
from azure.purview.catalog import PurviewCatalogClient from azure.identity import ClientSecretCredential from azure.core.exceptions import HttpResponseError
Créez une fonction pour obtenir les informations d’identification permettant d’accéder à votre compte Microsoft Purview et instancier le client de catalogue.
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)
Configurez vos critères de recherche et vos mots clés dans le corps d’entrée :
keywords = "keywords you want to search" body_input={ "keywords": keywords }
Ici, vous spécifiez uniquement des mots clés, mais gardez à l’esprit que vous pouvez ajouter de nombreux autres champs pour spécifier davantage votre requête.
Recherchez dans le catalogue :
try: response = client_catalog.discovery.query(search_request=body_input) print(response) except HttpResponseError as e: print(e)
Code complet
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)
Supprimer une source de données
Dans cette section, vous allez apprendre à supprimer la source de données que vous avez inscrite précédemment. Cette opération est assez simple et s’effectue avec le client d’analyse.
Importez le client d’analyse . Incluez également l’erreur HTTPResponse et ClientSecretCredential.
from azure.purview.scanning import PurviewScanningClient from azure.identity import ClientSecretCredential from azure.core.exceptions import HttpResponseError
Créez une fonction pour obtenir les informations d’identification pour accéder à votre compte Microsoft Purview et instancier le client d’analyse.
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)
Supprimez la source de données :
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)
Code complet
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)