SharePoint Online

SharePoint Online に接続して、指定した SharePoint ライブラリから Foundry にファイルをインポートします。

対応する機能

データモデル

このコネクターは、あらゆる種類のファイルを Foundry データセットに転送できます。ファイル形式は保持され、転送中も転送後もスキーマは適用されません。出力データセットに必要なスキーマを適用するか、下流のトランスフォームを作成してデータにアクセスします。

パフォーマンスと制限事項

転送可能なファイルのサイズに制限はありません。ただし、ネットワークの問題により、大規模な転送が失敗する場合があります。特に、実行に2日を超える時間がかかる Foundry の同期は中断されます。ネットワークの問題を避けるため、ファイルサイズを小さくし、同期の実行ごとに取り込むファイル数を制限することを推奨します。同期は、頻繁に実行するようにスケジュール設定できます。

セットアップ

1. Data Connection アプリケーションを開き、画面の右上にある + 新規ソースを選択します。
2. 利用可能なコネクターの種類から SharePoint Online を選択します。
3. 追加の設定プロンプトに従い、以下のセクションの情報を使用してコネクターのセットアップを進めます。

Foundry でのコネクターのセットアップについて、詳しくはこちらをご覧ください。

認証

SharePoint Online ソースの認証には、Microsoft Entra ID（旧称 Azure Active Directory）のアプリケーションが必要です。Entra ID 管理者でない場合は、IT 部門に連絡してアクセス権をリクエストしてください。

Azure アプリケーションの資格情報にアクセスするには、まず以下の手順を実行します。

1. Microsoft のドキュメント ↗ の手順に従って、Azure でアプリケーションを登録します。
   • ステップ5では、この組織ディレクトリ内のアカウントのみを選択し、リダイレクト URL（任意） はスキップします。
2. 登録が完了したら、クライアントIDとテナントIDを控えます。

次に、利用可能な2つの認証方法から選択します。

• クライアント資格情報：すべての SharePoint サイトに対する幅広いアクセスが必要な場合に推奨します。
• ユーザー名/パスワード：アクセスを1つまたは少数の SharePoint サイトに制限する場合に推奨します。

クライアント資格情報

Microsoft Entra 管理センターで、以下の手順を実行します。

1. 左側のサイドバーにある API のアクセス許可に移動します。

2. アクセス許可の追加を選択します。

3. Microsoft Graph を選択します。

4. アプリケーションの権限を選択します。

   • アプリケーションですべての SharePoint サイトを読み取る場合は、Sites.Read.All を追加します。
     • エクスポートタスクを設定する予定がある場合は、代わりに Sites.ReadWrite.All を使用します。
   • アプリケーションで選択した SharePoint サイトを読み取る場合は、Sites.Selected を追加します。

5. Entra 管理者の場合は、[tenant] に管理者の同意を与えますを選択します。

6. 上記で Sites.Selected を追加した場合は、アプリケーションを特定のサイトに追加 ↗します。

   • "roles" 配列パラメーターで利用可能な選択肢は、"write" と "read" のいずれか、または両方です。SharePoint サイトからファイルを取り込むには、"read" で十分です。
   • 適切な認証を使用して POST リクエストを簡単に送信するには、Graph Explorer ↗ を使用します。
   • https://graph.microsoft.com/v1.0/sites/[tenantName]:/sites/[siteName]（例：https://graph.microsoft.com/v1.0/sites/contoso.sharepoint.com:/sites/mySite）に GET リクエストを送信すると、サイトのメタデータを取得できます。このリクエストは、サイトコレクションのホスト名、サイトコレクションの一意の ID、サイトの一意の ID を組み合わせた ID を返します。中央の値が、アクセス許可の POST リクエストを実行するために必要な siteId です。

7. クライアントシークレットを生成します。↗。

Data Connection で、以下のソース設定を行います。

ユーザー名/パスワード

ユーザー名/パスワードのフローでは、Microsoft 365 にサインインできるユーザーアカウントを作成します。Graph API は、ユーザー名/パスワードの認証方法での2要素認証には対応していません。このため、32文字以上のランダムに生成されたパスワードを作成することを強く推奨します。

Entra 管理センターで、以下の手順を実行します。

1. 左側のサイドバーにある API のアクセス許可に移動します。
2. アクセス許可の追加を選択します。
3. Microsoft Graph を選択します。
4. 委任されたアクセス許可を選択します。
5. Sites.Read.All アクセス許可を追加します。
   • エクスポートタスクを設定する予定がある場合は、代わりに Sites.ReadWrite.All を使用します。
6. Azure 管理者の場合は、[tenant] に管理者の同意を与えますを選択します。
7. 左側のサイドバーにある認証に移動します。
8. パブリッククライアントフローを許可するを「はい」に変更します。
9. Microsoft Entra ID で、32文字以上のランダムに生成されたパスワードを持つユーザーを作成します。
10. そのユーザーに読み取りまたは書き込みを許可する SharePoint サイトに、そのユーザーを追加します。

Data Connection で、以下のソース設定を行います。

SharePoint アドインの XML ベースのアクセス許可管理

SharePoint アドインの承認と認証 ↗を使用し、その SharePoint アドインがアクセス許可管理に XML を使用している場合は、SharePoint への接続時にアクセスの問題が発生しないよう、スコープ URI に正しいスコープが設定されていることを確認してください。

以下の手順に従って、正しいスコープを確認し、設定します。

1. SharePoint アドインのアクセス許可設定が含まれる AppManifest.xml ファイルを見つけます。
2. AppManifest.xml ファイル内で、次のようなスコープ URI を特定します。

<AppPermissionRequests AllowAppOnlyPolicy="true"> <AppPermissionRequest Scope="http://sharepoint/content/sitecollection/web" Right="FullControl" /> </AppPermissionRequests>。

3. スコープの値（この例では http://sharepoint/content/sitecollection/web）が、接続先の SharePoint サイトと一致することを確認します。一致しない場合は、スコープの値を適切に調整します。

ネットワーク

以下のテーブルに、ソースの実行に必要なアクセス先ドメインを示します。

各ドメインに対応するエグレスポリシーを追加します。ソースがオンプレミスでホストされており、Foundry から直接アクセスできない場合は、代わりに Agent プロキシエグレスポリシーを使用します。Agent ホスト自体も、記載されたドメインにアクセスできる必要があります。詳細は、Agent をプロキシとして使用するを参照してください。

SharePoint Online コネクターには、ポート443で以下のドメインへのネットワークアクセスが必要です。

GovCloud の SharePoint インスタンスを使用している場合は、代わりにポート443で以下のドメインを使用します。

設定オプション

SharePoint Online コネクターでは、以下の設定オプションを利用できます。

SharePoint Online からデータを同期する

SharePoint Online コネクターは、ファイルベースの同期インターフェースを使用します。

SharePoint Online にデータをエクスポートする

SharePoint サイトにエクスポートするには、まず SharePoint Online コネクターのエクスポートを有効にします。次に、新しいエクスポートを作成します。

エクスポートの設定オプション

コードで SharePoint ソースを使用する

以下の例では、外部トランスフォームで SharePoint 用 Python クライアント ↗ Office365-REST-Python-Client を使用して、SharePoint ソースにファイルをアップロードする方法を示します。この例では、クライアント証明書認証を使用します。

SharePoint のその他の例 ↗も参照してください。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
from pyspark.sql import DataFrame
from transforms.api import Input, Output, transform, lightweight
from transforms.external.systems import external_systems, Source
import pandas as pd
import polars as pl
from office365.sharepoint.client_context import ClientContext

@lightweight
@external_systems(
    sharepoint_source=Source("<source_rid>")
)
@transform(
    output=Output("<dataset_rid>"),
    input_df=Input("<dataset_rid>"), # SharePoint にエクスポートするファイルの一覧を含むデータセット
)
def compute(ctx, input_df: DataFrame, output, sharepoint_source) -> DataFrame:

    # 1. クライアント証明書認証を使用して SharePoint に接続します。
    client = ClientContext("<sharepoint_url>").with_client_certificate(
        tenant="<tenant_id>",
        client_id="<client_id>",
        thumbprint="<thumbprint>",
        private_key=sharepoint_source.get_secret("clientSecret"),
    )

    current_web = client.web
    client.load(current_web)
    client.execute_query()

    target_folder = client.web.lists.get_by_title("<document_library_name>").root_folder

    # 2 input_df からファイルをアップロードし、URL をデータセットに保存します
    upload_urls = []
    fs = input_df.filesystem()
    input_files = fs.ls()
    for f in input_files:
        with fs.open(f.path) as fileobj:
            uploaded_file = target_folder.upload_file(f.path, fileobj).execute_query()
            upload_urls.append({'file_name': f.path, 'upload_url': uploaded_file.serverRelativeUrl})


    # 3. アップロード先の URL を含むデータセットを返します
    output.write_table(pl.from_pandas(pd.DataFrame.from_records(upload_urls)))

SharePoint リストを取り込む

SharePoint Online コネクターは、ファイルベースの取り込みのみをサポートします。SharePoint リストからデータを取り込むには、Microsoft Graph API を使用する外部トランスフォームを使用します。以下のヘルパークラスは OAuth2 認証を処理し、自動ページネーションでリストとリスト項目を取得するメソッドを提供します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
import requests
import logging
from typing import Optional, Dict, List
from urllib.parse import urlparse


class SharePointListReader:
    """
    Microsoft Graph API 経由で SharePoint リストを読み込むクライアントです。

    このクラスは OAuth2 認証を処理し、次の操作を行うメソッドを提供します。
    - SharePoint サイトからすべてのリストを取得します
    - 自動ページネーションを使用して特定のリストから項目を取得します

    Args:
        tenant_id: Azure AD テナントID
        client_id: Azure AD アプリケーション（クライアント）ID
        client_secret: Azure AD アプリケーションのクライアントシークレット
        site_url: SharePoint サイトの完全な URL（例：https://contoso.sharepoint.com/sites/mysite）
        logger: カスタムロガーのインスタンス（任意）
    """

    def __init__(
        self,
        tenant_id: str,
        client_id: str,
        client_secret: str,
        site_url: str,
        logger: Optional[logging.Logger] = None
    ):
        self.tenant_id = tenant_id
        self.client_id = client_id
        self.client_secret = client_secret
        self.site_url = site_url.rstrip("/")
        self.base_url = "https://graph.microsoft.com/v1.0"

        self.access_token: Optional[str] = None
        self.site_id: Optional[str] = None
        self.logger = logger or self._setup_default_logger()

    def _setup_default_logger(self) -> logging.Logger:
        """コンソール出力を行うデフォルトのロガーを設定します。"""
        logger = logging.getLogger(__name__)
        if not logger.handlers:
            handler = logging.StreamHandler()
            formatter = logging.Formatter('%(levelname)s: %(message)s')
            handler.setFormatter(formatter)
            logger.addHandler(handler)
            logger.setLevel(logging.INFO)
        return logger

    def get_access_token(self) -> bool:
        """
        Azure AD から OAuth2 アクセストークンを取得します。

        Returns:
            トークンの取得に成功した場合は True、それ以外の場合は False
        """
        token_url = f"https://login.microsoftonline.com/{self.tenant_id}/oauth2/v2.0/token"
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "https://graph.microsoft.com/.default",
        }

        try:
            response = requests.post(token_url, data=payload)
            response.raise_for_status()
            token_data = response.json()
            self.access_token = token_data["access_token"]

            expires_in = token_data.get("expires_in", 3600)
            self.logger.info(f"Authentication successful (expires in {expires_in}s)")
            return True

        except requests.exceptions.RequestException as e:
            self.logger.error(f"Authentication failed: {e}")
            return False

    def _make_graph_request(self, url: str, params: Optional[Dict] = None) -> Optional[Dict]:
        """
        Microsoft Graph API への認証付き GET リクエストを実行します。

        Args:
            url: Graph API エンドポイントの完全な URL
            params: クエリパラメーター（任意）

        Returns:
            辞書形式の JSON レスポンス、失敗した場合は None
        """
        if not self.access_token and not self.get_access_token():
            return None

        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type": "application/json",
        }

        try:
            response = requests.get(url, headers=headers, params=params)
            response.raise_for_status()
            return response.json()

        except requests.exceptions.RequestException as e:
            self.logger.error(f"API request failed: {e}")
            if hasattr(e, 'response') and e.response is not None:
                self.logger.debug(f"Response details: {e.response.text}")
            return None

    def get_site_id(self) -> Optional[str]:
        """
        サイト URL から SharePoint サイトIDを取得します。

        Returns:
            サイトIDの文字列、取得に失敗した場合は None
        """
        if self.site_id:
            return self.site_id

        parsed = urlparse(self.site_url)
        hostname = parsed.hostname
        site_path = parsed.path.strip("/")

        url = f"{self.base_url}/sites/{hostname}:/{site_path}"
        data = self._make_graph_request(url)

        if data and "id" in data:
            self.site_id = data["id"]
            self.logger.debug(f"Site ID retrieved: {self.site_id}")
            return self.site_id

        self.logger.error("Failed to retrieve site ID")
        return None

    def get_all_lists(self) -> Optional[Dict]:
        """
        SharePoint サイトからすべてのリストを取得します。

        Returns:
            リストのメタデータを含む辞書、失敗した場合は None
        """
        site_id = self.get_site_id()
        if not site_id:
            return None

        url = f"{self.base_url}/sites/{site_id}/lists"
        data = self._make_graph_request(url)

        if data and "value" in data:
            self.logger.info(f"Found {len(data['value'])} lists in site")
            for lst in data["value"]:
                self.logger.info(f"  - {lst['name']} (ID: {lst['id']})")

        return data

    def get_all_list_items(self, list_id: str) -> Optional[List[Dict]]:
        """
        自動ページネーションを使用して SharePoint リストからすべての項目を取得します。

        Args:
            list_id: SharePoint リストの GUID

        Returns:
            項目の辞書を含むリスト、失敗した場合は None
        """
        site_id = self.get_site_id()
        if not site_id:
            return None

        all_items = []
        url = f"{self.base_url}/sites/{site_id}/lists/{list_id}/items"
        params = {"$expand": "fields", "$top": 5000}

        page_count = 0
        while url:
            current_params = None if "@odata.nextLink" in url else params
            data = self._make_graph_request(url, current_params)

            if not data or "value" not in data:
                break

            page_count += 1
            items_in_page = len(data["value"])
            all_items.extend(data["value"])

            self.logger.debug(f"Page {page_count}: retrieved {items_in_page} items")

            url = data.get("@odata.nextLink")
            params = None

        self.logger.info(f"Retrieved {len(all_items)} total items from list")
        return all_items

以下の例では、外部トランスフォームでこのクラスを使用して、SharePoint リストのデータを Foundry データセットに取り込む方法を示します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from transforms.api import Output, transform, lightweight
from transforms.external.systems import external_systems, Source
import polars as pl


@lightweight
@external_systems(
    sharepoint_source=Source("<source_rid>")
)
@transform(
    output=Output("<dataset_rid>"),
)
def compute(ctx, output, sharepoint_source):

    # 1. ソースの資格情報を使用して SharePoint リストリーダーを初期化します
    reader = SharePointListReader(
        tenant_id="<tenant_id>",
        client_id="<client_id>",
        client_secret=sharepoint_source.get_secret("clientSecret"),
        site_url="https://contoso.sharepoint.com/sites/mysite"
    )

    # 2. 特定のリストからすべての項目を取得します
    items = reader.get_all_list_items(list_id="<list_guid>")

    # 3. 各項目からフィールドを抽出し、出力データセットに書き込みます
    records = [item["fields"] for item in items if "fields" in item]
    output.write_table(pl.from_dicts(records))