教學課程：使用 Microsoft Purview Python SDK

本教學課程將介紹如何使用 Microsoft Purview Python SDK。 您可以使用 SDK，以程式設計方式執行所有最常見的 Microsoft Purview 作業，而不是透過 Microsoft Purview 治理入口網站。

在本教學課程中，您將瞭解 SDK 如何：

• 授與以程序設計方式使用 Purview 的必要許可權Microsoft
• 在 Purview 中將 Blob 記憶體容器註冊為數據源Microsoft
• 定義和執行掃描
• 搜尋目錄
• 刪除數據源

必要條件

在本教學課程中，您將需要：

• Python 3.6 或更高版本
• 作用中的 Azure 訂用帳戶。 如果您沒有帳戶，可以免費建立一個。
• 與您的訂用帳戶相關聯的 Microsoft Entra 租使用者。
• Azure 記憶體帳戶。 如果您還沒有，您可以 遵循我們的快速入門指南來建立一個。
• Microsoft Purview 帳戶。 如果您還沒有，您可以 遵循我們的快速入門指南來建立一個。
• 具有客戶端密碼的服務主體。

重要事項

針對這些腳本，端點值會根據您使用的 Purview 入口網站Microsoft而有所不同。 傳統 Microsoft Purview 治理入口網站的端點：purview.azure.com/ 新Microsoft Purview 入口網站的端點：purview.microsoft.com/

因此，如果您使用新的入口網站，您的端點值會類似：“https://consotopurview.scan.purview.microsoft.com"

授與 Microsoft 記憶體帳戶的 Purview 存取權

在能夠掃描記憶體帳戶的內容之前，您必須為 Microsoft 提供正確的角色。

1. 透過 Azure 入口網站 移至您的記憶體帳戶。

2. 選 存取控制 (IAM) 。

3. 選取 [新增] 按鈕，然後選取 [ 新增角色指派]。

   

4. 在下一個視窗中，搜尋 記憶體 Blob 讀取器 角色並加以選取：

   

5. 然後，移至 [ 成員] 索引 標籤，然後選 取 [選取成員]：

   

6. 右側會出現新的窗格。 搜尋並選取現有 Microsoft Purview 實例的名稱。

7. 然後，您可以選取 [ 檢閱 + 指派]。

Microsoft Purview 現在具有掃描 Blob 記憶體所需的讀取許可權。

將您Microsoft Purview 帳戶的存取權授與您的應用程式

1. 首先，您需要服務主體的用戶端標識碼、租使用者標識碼和客戶端密碼。 若要尋找此資訊，請選取您的 Microsoft Entra ID。

2. 然後，選取 [應用程式註冊]。

3. 選取您的應用程式，並找出必要的資訊：

   • 名稱

   • 用戶端識別碼 (或應用程式標識碼)

   • 租用戶標識碼 (或目錄標識碼)

     

   • 用戶端密碼

     

4. 您現在必須將相關Microsoft Purview 角色提供給服務主體。 若要這樣做，請存取您的 Microsoft Purview 實例。 選 取 [開啟Microsoft Purview 治理入口網站 ]，或 直接開啟 Microsoft Purview 的治理入口網站 ，然後選擇您部署的實例。

5. 在 Microsoft Purview 治理入口網站中，依序選取 [數據對應] 和 [ 集合]：

   

6. 選取您要使用的集合，然後移至 [ 角色指派] 索引標籤 。在下列角色中新增服務主體：

   • 集合管理員
   • 數據源管理員
   • 數據編者
   • 數據讀取器

7. 針對每個角色，選取 [ 編輯角色指派] 按鈕，然後選取您要新增服務主體的角色。 或選取每個角色旁邊的 [ 新增 ] 按鈕，並搜尋其名稱或用戶端標識符來新增服務主體，如下所示：

   

安裝 Python 套件

1. 開啟新的命令提示字元或終端機
2. 安裝用於驗證的 Azure 身分識別套件：

   pip install azure-identity
   

3. 安裝 Microsoft Purview 掃描用戶端套件：

   pip install azure-purview-scanning
   

4. 安裝 Microsoft Purview 管理用戶端套件：

   pip install azure-purview-administration
   

5. 安裝 Microsoft Purview 用戶端套件：

   pip install azure-purview-catalog
   

6. 安裝 Microsoft Purview 帳戶套件：

   pip install azure-purview-account
   

7. 安裝 Azure Core 套件：

   pip install azure-core
   

建立 Python 腳本檔案

建立純文本檔，並將其儲存為後綴為 .py 的 Python 腳本。 例如：tutorial.py。

具現化掃描、目錄和管理用戶端

在本節中，您將瞭解如何具現化：

• 掃描用戶端適用於註冊數據源、建立和管理掃描規則、觸發掃描等等。
• 目錄客戶端有助於透過搜尋、流覽探索到的資產、識別數據的敏感度等與目錄互動。
• 系統管理客戶端適用於與 Microsoft Purview 資料對應 本身互動，適用於清單集合等作業。

首先，您需要向您的 Microsoft Entra ID 進行驗證。 為此，您將使用 您建立的客戶端密碼。

1. 從必要的匯入語句開始：我們的三個用戶端、認證語句和 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. 在程式代碼指定以下資訊：

   • 用戶端識別碼 (或應用程式標識碼)
   • 租用戶標識碼 (或目錄標識碼)
   • 用戶端密碼

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

3. 指定您的端點：

   重要事項

   您的端點值會根據您使用的 Purview 入口網站Microsoft而有所不同。 傳統 Microsoft Purview 治理入口網站的端點： https://{your_purview_account_name}.purview.azure.com/ 新Microsoft Purview 入口網站的端點： https://api.purview-service.microsoft.com

   掃描傳統 Microsoft Purview 治理入口網站的端點： https://{your_purview_account_name}.scan.purview.azure.com/ 新Microsoft Purview 入口網站的端點： https://api.scan.purview-service.microsoft.com

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

4. 您現在可以具現化這三個用戶端：

   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
   

我們的許多腳本都會從這些相同的步驟開始，因為我們需要這些客戶端與帳戶互動。

註冊數據源

在本節中，您將註冊 Blob 記憶體。

1. 如同我們在上一節中所討論，首先您將匯入存取 Microsoft Purview 帳戶所需的用戶端。 也請匯入 Azure 錯誤回應套件，以便進行疑難解答，並匯入 ClientSecretCredential 來建構您的 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. 依照本指南收集記憶體帳戶的資源標識碼： 取得記憶體帳戶的資源標識碼。

3. 然後，在 Python 檔案中定義下列資訊，以便能夠以程式設計方式註冊 Blob 記憶體：

   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. 提供您要在其中註冊 Blob 記憶體的集合名稱。 (它應該是您稍早套用許可權的相同集合。如果不是，請先將許可權套用至此集合。) 如果是根集合，請使用與您Microsoft Purview 實例相同的名稱。

   collection_name = "<name of your collection>"
   

5. 建立函式來建構認證，以存取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. Microsoft Purview 資料對應 中的所有集合都有易記名稱和名稱。

   • 易 記名稱 是您在集合上看到的名稱。 例如：Sales。
   • 除了根集合) 之外，所有集合 (的 名稱 都是數據對應所指派的六個字元名稱。

   Python 需要這個六個字元的名稱來參考任何子集合。 若要自動將易 記名稱 轉換為文稿中所需的六個字元集合名稱，請新增下列程式代碼區塊：

   重要事項

   您的端點值會根據您使用的 Purview 入口網站Microsoft而有所不同。 傳統 Microsoft Purview 治理入口網站的端點：purview.azure.com/ 新Microsoft Purview 入口網站的端點：purview.microsoft.com/

   因此，如果您使用新的入口網站，您的端點值會類似：“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. 針對這兩個用戶端，以及根據作業，您也需要提供輸入本文。 若要註冊來源，您必須提供數據源註冊的輸入主體：

   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. 現在您可以呼叫Microsoft Purview 用戶端並註冊數據源。

   重要事項

   您的端點值會根據您使用的 Purview 入口網站Microsoft而有所不同。 傳統 Microsoft Purview 治理入口網站的端點： https://{your_purview_account_name}.purview.azure.com/ 新Microsoft Purview 入口網站的端點： https://api.purview-service.microsoft.com

   如果您使用傳統入口網站，您的端點值會是： https://{your_purview_account_name}.scan.purview.azure.com 如果您使用新的入口網站，您的端點值會是： 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)
   

註冊程式成功時，您可以看到來自用戶端的擴充主體回應。

在下列各節中，您將掃描已註冊的數據源，並搜尋目錄。 這些腳本的結構與此註冊腳本類似。

完整程序代碼

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)

掃描數據源

掃描資料來源有兩個步驟：

1. 建立掃描定義
2. 觸發掃描執行

在本教學課程中，您將使用 Blob 記憶體容器的預設掃描規則。 不過，您也可以使用 Microsoft Purview 掃描用戶端，以程式設計方式建立自定義掃描規則。

現在讓我們掃描您在上面註冊的數據源。

1. 新增 import 語句以產生 唯一標識符、呼叫 Microsoft Purview 掃描用戶端、Microsoft Purview 管理客戶端、能夠進行疑難解答的 Azure 錯誤回應套件，以及收集 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. 使用您的認證建立掃描用戶端：

   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. 新增程式代碼以收集集合的內部名稱。 (如需詳細資訊，請參閱上一節) ：

   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. 然後，建立掃描定義：

   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. 現在已定義掃描，您可以使用唯一識別子觸發掃描執行：

   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)
   

完整程序代碼

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)

搜尋目錄

掃描完成後，很可能已探索資產，甚至分類。 此程式可能需要一些時間才能在掃描之後完成，因此您可能需要先等候再執行下一部分的程序代碼。 等候掃描顯示完成，資產會出現在 Microsoft Purview 整合式目錄 中。

資產準備就緒后，您可以使用 Microsoft Purview 目錄用戶端來搜尋整個目錄。

1. 這次您需要匯 入目錄 用戶端，而不是掃描用戶端。 也包括 HTTPResponse 錯誤和 ClientSecretCredential。

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

2. 建立函式以取得認證以存取您的 Microsoft Purview 帳戶，並具現化目錄用戶端。

   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. 在輸入主體中設定搜尋準則和關鍵詞：

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

   在這裡，您只會指定關鍵詞，但請記住 ，您可以新增許多其他欄位來進一步指定您的查詢。

4. 搜尋目錄：

   try:
       response = client_catalog.discovery.query(search_request=body_input)
       print(response)
   except HttpResponseError as e:
       print(e)
   

完整程序代碼

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)

刪除數據源

在本節中，您將瞭解如何刪除您稍早註冊的數據源。 這項作業相當簡單，而且會使用掃描用戶端來完成。

1. 匯入 掃描 用戶端。 也包括 HTTPResponse 錯誤和 ClientSecretCredential。

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

2. 建立函式以取得認證以存取您的 Microsoft Purview 帳戶，並具現化掃描用戶端。

   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. 刪除資料來源：

       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)
   

完整程序代碼

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)

後續步驟

深入瞭解 Python Microsoft Purview 掃描客戶端深入瞭解 Python Microsoft Purview 目錄用戶端