Share via


チュートリアル: Microsoft Purview Python SDK を使用する

このチュートリアルでは、Microsoft Purview Python SDK の使用について説明します。 SDK を使用すると、Microsoft Purview ガバナンス ポータルではなく、最も一般的な Microsoft Purview 操作をすべてプログラムで実行できます。

このチュートリアルでは、SDK で次の操作を行う方法について説明します。

  • Microsoft Purview をプログラムで操作するために必要な権限を付与する
  • Microsoft Purview で Blob Storage コンテナーをデータ ソースとして登録する
  • スキャンを定義して実行する
  • カタログを検索する
  • データ ソースを削除する

前提条件

このチュートリアルでは、次のものが必要です。

重要

これらのスクリプトでは、使用している Microsoft Purview ポータルによってエンドポイント値が異なります。 従来の Microsoft Purview ガバナンス ポータルのエンドポイント: 新しい Microsoft Purview ポータルの purview.azure.com/ エンドポイント: purview.microsoft.com/

そのため、新しいポータルを使用している場合、エンドポイント値は次のようになります。https://consotopurview.scan.purview.microsoft.com"

Microsoft Purview にストレージ アカウントへのアクセス権を付与する

ストレージ アカウントのコンテンツをスキャンする前に、Microsoft Purview に適切なロールを付与する必要があります。

  1. Azure portalからストレージ アカウントに移動します。

  2. [Access Control (IAM)] を選択します。

  3. [追加] ボタンを選択し、[ ロールの割り当ての追加] を選択します。

    [追加] ボタンが選択され、[ロールの割り当てを追加] が選択されているストレージ アカウントの [Access Control] メニューのスクリーンショット。

  4. 次のウィンドウで、 ストレージ BLOB 閲覧者 ロールを検索して選択します。

    使用可能なロールの一覧から [ストレージ BLOB データ閲覧者] が選択されている [ロールの割り当ての追加] メニューのスクリーンショット。

  5. 次に、[ メンバー ] タブに移動し、[ メンバーの選択] を選択します。

    [+ メンバーの選択] ボタンが選択されている [ロールの割り当ての追加] メニューのスクリーンショット。

  6. 右側に新しいウィンドウが表示されます。 既存の Microsoft Purview インスタンスの名前を検索して選択します。

  7. その後、[ 確認と割り当て] を選択できます。

Microsoft Purview には、Blob Storage をスキャンするために必要な読み取り権限が付与されました。

アプリケーションに Microsoft Purview アカウントへのアクセス権を付与する

  1. まず、サービス プリンシパルのクライアント ID、テナント ID、クライアント シークレットが必要です。 この情報を見つけるには、 Azure Active Directory を選択します。

  2. 次に、[アプリの登録] を選択します。

  3. アプリケーションを選択し、必要な情報を見つけます。

  4. これで、関連する Microsoft Purview ロールをサービス プリンシパルに付与する必要があります。 これを行うには、Microsoft Purview インスタンスにアクセスします。 [ Microsoft Purview ガバナンス ポータルを開く ] を選択するか 、Microsoft Purview のガバナンス ポータルを直接 開き、デプロイしたインスタンスを選択します。

  5. Microsoft Purview ガバナンス ポータルで、[ データ マップ]、[ コレクション] の順に選択します。

    Microsoft Purview ガバナンス ポータルの左側のメニューのスクリーンショット。[データ マップ] タブが選択され、[コレクション] タブが選択されます。

  6. 使用するコレクションを選択し、[ ロールの割り当て ] タブに移動します。次のロールにサービス プリンシパルを追加します。

    • コレクション管理者
    • データ ソース管理者
    • データ キュレーター
    • データ リーダー
  7. ロールごとに、[ ロールの割り当ての編集 ] ボタンを選択し、サービス プリンシパルを追加するロールを選択します。 または、各ロールの横にある [ 追加] ボタンを選択し、次に示すようにサービス プリンシパルの名前またはクライアント ID を検索してサービス プリンシパルを追加します。

    Microsoft Purview ガバナンス ポータルのコレクションの下にある [ロールの割り当て] メニューのスクリーンショット。[コレクション管理者] タブの横にある [ユーザーの追加] ボタンが選択されます。[コレクション管理者の追加と削除] ウィンドウが表示され、テキスト ボックスにサービス プリンシパルが検索されます。

Python パッケージをインストールする

  1. 新しいコマンド プロンプトまたはターミナルを開く
  2. 認証用の Azure ID パッケージをインストールします。
    pip install azure-identity
    
  3. Microsoft Purview Scanning Client パッケージをインストールします。
    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 データ マップ自体と対話する場合に役立ちます。

まず、Azure Active Directory に対して認証する必要があります。 このためには、 作成したクライアント シークレットを使用します。

  1. 必要な import ステートメント(3 つのクライアント、credentials ステートメント、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. コードで次の情報を指定します。

    • クライアント ID (またはアプリケーション ID)
    • テナント ID (またはディレクトリ ID)
    • クライアント シークレット
    client_id = "<your client id>" 
    client_secret = "<your client secret>"
    tenant_id = "<your tenant id>"
    
  3. エンドポイントを指定します。

    重要

    エンドポイントの値は、使用している Microsoft Purview ポータルによって異なります。 クラシック Microsoft Purview ガバナンス ポータルのエンドポイント: 新しい Microsoft Purview ポータルのエンドポイント: https://{your_purview_account_name}.purview.azure.com/https://api.purview-service.microsoft.com

    クラシック Microsoft Purview ガバナンス ポータルのエンドポイントをスキャンする: 新しい Microsoft Purview ポータルのエンドポイント: https://{your_purview_account_name}.scan.purview.azure.com/https://api.scan.purview-service.microsoft.com

    purview_endpoint = "<endpoint>"
    
    purview_scan_endpoint = "<scan endpoint>"
    
  4. これで、次の 3 つのクライアントをインスタンス化できます。

    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 Storage を登録します。

  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. このガイドに従って、ストレージ アカウントの リソース ID を収集します。ストレージ アカウントのリソース ID を取得します。

  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 です。
    • (ルート コレクションを除く) すべてのコレクションの 名前 は、データ マップによって割り当てられた 6 文字の名前です。

    Python では、サブ コレクションを参照するために、この 6 文字の名前が必要です。 フレンドリ名をスクリプトに必要な 6 文字のコレクション名に自動的に変換するには、次のコード ブロックを追加します。

    重要

    エンドポイントの値は、使用している Microsoft Purview ポータルによって異なります。 従来の Microsoft Purview ガバナンス ポータルのエンドポイント: 新しい Microsoft Purview ポータルの purview.azure.com/ エンドポイント: 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 クライアントを呼び出して、データ ソースを登録できます。

    重要

    エンドポイントの値は、使用している Microsoft Purview ポータルによって異なります。 クラシック Microsoft Purview ガバナンス ポータルのエンドポイント: 新しい Microsoft Purview ポータルのエンドポイント: https://{your_purview_account_name}.purview.azure.com/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)

データ ソースをスキャンする

データ ソースのスキャンは、次の 2 つの手順で実行できます。

  1. スキャン定義を作成する
  2. スキャン実行をトリガーする

このチュートリアルでは、Blob Storage コンテナーの既定のスキャン 規則を使用します。 ただし、 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. スキャンが定義されたので、一意の ID でスキャン実行をトリガーできます。

    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 Catalog クライアントを使用してカタログ全体を検索できます。

  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=purview_endpoint, 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)

次の手順