Keyboard shortcuts

Press ← or → to navigate between chapters

Press ? to show this help

Press Esc to hide this help

SharePoint Online 파일과 목록 데이터

SharePoint Online 커넥터는 지정한 문서 라이브러리의 파일을 Foundry로 가져옵니다. 원천 탐색, 대량 가져오기, 증분 수집과 파일 내보내기는 일반 제공(GA)이며 Export tasks는 Sunset 단계입니다.

파일 형식을 유지하고 스키마는 자동 적용하지 않습니다. 파일 크기 자체의 한도는 없지만 이틀 넘게 실행되는 Foundry 동기화는 중단됩니다. 파일을 작게 나누고 한 번에 가져올 수를 제한하거나 실행 주기를 짧게 구성합니다.

온프레미스 SharePoint 서버에는 이 커넥터를 사용할 수 없습니다. 온프레미스 연결은 REST API 소스로 구성합니다.

앱 등록

Data Connection의 +New Source에서 SharePoint Online 소스를 선택합니다. 인증에는 Microsoft Entra ID 앱이 필요합니다. Entra 관리자 권한이 없으면 조직 IT 담당자에게 앱 구성을 요청합니다.

앱 등록 시 Accounts in this organizational directory only를 선택하고 Redirect URL은 지정하지 않습니다. 등록 후 client ID와 tenant ID를 보관합니다. 인증 방식은 애플리케이션의 client credentials 또는 사용자 이름과 비밀번호 중 선택합니다.

애플리케이션 권한

Entra 관리 센터에서 API Permissions, Add a Permission, Microsoft Graph, Application Permissions를 선택합니다.

요구 범위권한
모든 사이트 읽기Sites.Read.All
모든 사이트에 쓰는 내보내기Sites.ReadWrite.All
지정한 사이트만 접근Sites.Selected와 사이트별 앱 권한

관리자가 Grant admin consent for [tenant]를 수행합니다. Sites.Selected는 그것만 부여해서 끝나지 않으며 앱을 각 사이트에 추가해야 합니다. 사이트별 roles에는 read, write를 지정할 수 있고 파일 수집에는 read로 충분합니다.

Graph Explorer 등에서 사이트 메타데이터를 확인할 때 사용하는 URL 형식은 다음과 같습니다.

https://graph.microsoft.com/v1.0/sites/[tenantName]:/sites/[siteName]
https://graph.microsoft.com/v1.0/sites/contoso.sharepoint.com:/sites/mySite

두 번째 주소는 예시입니다. 응답의 사이트 ID는 사이트 모음 호스트, 사이트 모음 고유 ID와 사이트 고유 ID를 결합한 값입니다. 뒤의 Graph 목록 예제는 응답의 전체 id를 후속 사이트 경로에 사용합니다.

앱의 client secret을 생성하고 Foundry에 다음 필수 항목을 입력합니다.

항목의미
Azure Client ID앱 등록의 Application ID입니다.
Azure Tenant IDEntra 인스턴스의 고유 식별자입니다.
Client secret앱 등록에서 생성한 비밀 값입니다.

사용자 이름과 비밀번호 인증

이 방식은 Microsoft 365에 로그인할 수 있는 사용자를 이용합니다. Graph의 이 사용자 이름/비밀번호 흐름은 2단계 인증을 지원하지 않습니다. 최소 32자의 임의 생성 비밀번호를 사용하는 전용 계정을 준비하고, MFA 요구와 양립 가능한지 확인해야 합니다.

  1. 앱의 API Permissions에서 Microsoft Graph의 Delegated Permissions를 선택합니다.
  2. 읽기에는 Sites.Read.All, 내보내기에는 Sites.ReadWrite.All을 추가합니다.
  3. 관리자가 tenant에 대한 동의를 부여합니다.
  4. Authentication의 Allow public client flows를 Yes로 설정합니다.
  5. 최소 32자 임의 비밀번호를 가진 사용자를 만들고 접근할 SharePoint 사이트에 추가합니다.

Foundry의 필수 항목은 Azure Client ID, 사용자 이메일 주소인 Username, 생성한 Password입니다. 사용자에게 부여한 사이트 접근 범위 안에서 사용합니다.

기존 Add-in XML 권한

SharePoint Add-in의 XML로 권한을 관리하는 구성에서는 AppManifest.xml의 Scope URI가 연결 대상 범위와 맞는지 확인합니다. 다음은 site collection의 web 범위와 FullControl을 표현한 형식 예시이며 모든 수집에 FullControl이 필요하다는 뜻은 아닙니다.

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

네트워크와 라이브러리 위치

일반 환경에서는 login.microsoftonline.com, graph.microsoft.com, 실제 SharePoint 호스트(예: contoso.sharepoint.com)의 443 포트를 허용합니다. GovCloud에서는 대신 login.microsoftonline.us, graph.microsoft.us, 실제 GovCloud SharePoint 호스트(예: contoso.sharepoint.us)의 443 포트를 허용합니다.

외부 통신 정책을 대상별로 추가합니다. 직접 도달할 수 없는 경로에는 에이전트 프록시 정책을 사용하며 에이전트 호스트도 대상에 접근할 수 있어야 합니다.

소스 항목필수 여부의미
SharePoint Library URL필수특정 문서 라이브러리를 가리켜야 합니다. 형식은 https://[tenant].sharepoint.com/sites/[site]/[library]입니다.
Credentials settings필수선택한 인증 방식을 구성합니다.
Proxy settings선택프록시를 구성합니다.

사이트 하나에는 여러 라이브러리가 있을 수 있으므로 사이트 루트 주소만 지정하지 않습니다.

수집과 내보내기

수집에는 파일 기반 동기화를 사용합니다. 내보내기를 활성화한 뒤 새 파일 내보내기를 생성할 수 있습니다. 필수 Directory path의 기본값은 /이며 라이브러리 안의 대상 폴더를 가리킵니다.

최종 경로는 <SharePoint Library URL>/<Directory Path>/<Exported File Path>입니다.

인증서로 파일 업로드

다음 외부 변환은 Office365-REST-Python-Client와 client certificate 인증으로 파일을 업로드합니다. 앱의 인증서 등록, tenant, client ID와 thumbprint를 준비합니다. 업로드용 소스의 clientCertificatePrivateKey 비밀 정보에는 인증서의 private key를 저장합니다. OAuth client secret 문자열과 혼동하지 않습니다.

소스 RID, 서로 다른 입력 및 출력 데이터셋 RID, 사이트 URL과 문서 라이브러리 이름을 실제 값으로 바꿉니다. 입력은 파일 데이터셋이고 출력은 파일 이름과 업로드 URL의 대응 표입니다. 실행하면 SharePoint에 파일을 쓰므로 대상과 쓰기 권한을 확인합니다.

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("<output_dataset_rid>"),
    input_df=Input("<input_dataset_rid>"), # Dataset containing a list of files to export to SharePoint
)
def compute(ctx, input_df, output, sharepoint_source):

    # 1. Connect to SharePoint using client certificate authentication.
    client = ClientContext("<sharepoint_url>").with_client_certificate(
        tenant="<tenant_id>",
        client_id="<client_id>",
        thumbprint="<thumbprint>",
        private_key=sharepoint_source.get_secret("clientCertificatePrivateKey"),
    )

    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 Upload files from input_df, store URL in dataset
    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. Return dataset of uploaded URLs
    output.write_table(pl.from_pandas(pd.DataFrame.from_records(upload_urls)))

SharePoint Lists 수집

기본 커넥터는 파일 수집만 지원합니다. SharePoint Lists 데이터는 외부 변환에서 Microsoft Graph를 호출하여 가져옵니다.

다음 클래스는 client credentials 토큰을 발급받고 사이트 URL로 사이트 ID를 조회합니다. get_all_lists는 목록 메타데이터 응답을 반환합니다. get_all_list_items는 $expand=fields, $top=5000으로 항목을 요청하고 @odata.nextLink를 따라 항목 페이지를 수집합니다.

메서드는 실패하면 None을 반환합니다. 중간 페이지에서 실패한 결과를 완전한 목록으로 취급하지 않도록 호출 측이 이를 오류로 처리합니다. 토큰은 객체에 보관되지만 만료 시간을 이용한 자동 갱신은 구현하지 않으므로 장기간 사용하는 호출에서는 토큰 수명을 고려합니다. 이 예제의 인증과 Graph 호스트는 일반 상용 환경을 대상으로 합니다.

import requests
import logging
from typing import Optional, Dict, List
from urllib.parse import urlparse


class SharePointListReader:
    """
    Client for reading SharePoint lists via Microsoft Graph API.

    This class handles OAuth2 authentication and provides methods to:
    - Retrieve all lists from a SharePoint site
    - Fetch items from specific lists with automatic pagination

    Args:
        tenant_id: Azure AD tenant ID
        client_id: Azure AD application (client) ID
        client_secret: Azure AD application client secret
        site_url: Full SharePoint site URL (e.g., https://contoso.sharepoint.com/sites/mysite)
        logger: Optional custom logger instance
    """

    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:
        """Configure default logger with console output."""
        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:
        """
        Acquire OAuth2 access token from Azure AD.

        Returns:
            True if token was successfully acquired, False otherwise
        """
        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]:
        """
        Execute authenticated GET request to Microsoft Graph API.

        Args:
            url: Full Graph API endpoint URL
            params: Optional query parameters

        Returns:
            JSON response as dictionary, or None on failure
        """
        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]:
        """
        Retrieve SharePoint site ID from site URL.

        Returns:
            Site ID string, or None if retrieval fails
        """
        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]:
        """
        Retrieve all lists from the SharePoint site.

        Returns:
            Dictionary containing list metadata, or None on failure
        """
        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]]:
        """
        Retrieve all items from a SharePoint list with automatic pagination.

        Args:
            list_id: GUID of the SharePoint list

        Returns:
            List of item dictionaries, or None on failure
        """
        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:
                return None

            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

같은 모듈에 위 클래스를 두고 다음 변환에서 호출합니다. source의 clientSecret에는 이 client credentials 흐름에 사용할 실제 OAuth client secret을 저장합니다. 파일 업로드 예제의 인증서 private key와는 다른 값입니다.

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. Initialize the SharePoint List reader with credentials from the source
    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. Retrieve all items from a specific list
    items = reader.get_all_list_items(list_id="<list_guid>")

    if items is None:
        raise RuntimeError("SharePoint list retrieval failed")

    # 3. Extract the fields from each item and write to the output dataset
    records = [item["fields"] for item in items if "fields" in item]
    output.write_table(pl.from_dicts(records))

출력에는 각 항목의 fields를 사용하므로 필드 구조는 대상 목록에 따라 달라집니다. 오류 응답이나 일부 페이지만 받은 상태를 정상적인 전체 수집 결과로 출력하지 않습니다.