MSSharePointFetcher

Microsoft Graph API를 통해 SharePoint·OneDrive 항목의 전체 내용을 내려받아 ByteStream으로 돌려주는 컴포넌트예요.

출처: 문서

본문

MSSharePointFetcher는 Microsoft Graph API를 통해 Microsoft SharePoint·OneDrive 항목의 전체 내용을 내려받고, 다운스트림 변환기(converter)가 바로 쓸 수 있는 ByteStream 객체로 돌려줘요.

이 컴포넌트는 검색 스니펫과 메타데이터만 돌려주는 MSSharePointRetriever를 보완해요. Retriever의 documents(또는 web_url 목록)를 Fetcher로 연결하면 실제 내용을 내려받게 되죠. 각 검색 결과의 엔티티 유형에 따라 처리가 달라져요.

파일(driveItem)은 원본 바이트 그대로 내려받아요(PDF, DOCX 등).
목록 항목(listItem)은 항목의 열 값(fields)을 JSON ByteStream으로 돌려줘요.
SharePoint 페이지(sitePage)는 페이지의 web part로 만든 HTML ByteStream으로 돌려줘요.

각 ByteStream의 meta에는 url, file_name, content_type, 정규화된 entity_type(driveItem, listItem, sitePage)이 담겨요. 모든 것이 Microsoft Graph shares 엔드포인트(페이지는 Pages API 추가)로 해석되므로, Retriever가 이미 노출한 web_url만 있으면 돼요.

출력이 여러 유형이 섞인 ByteStream 목록이라, 다음 단계로는 보통 FileTypeRouter를 붙여 각 스트림을 알맞은 변환기(PyPDFToDocument, DOCXToDocument, HTMLToDocument, JSON 변환기)로 분배하는 방식이에요.

인증

Fetcher는 실행 입력으로 사용자별 access_token을 받아요. 토큰은 위임된(delegated) Microsoft Graph 권한을 지녀야 해요(예: 파일용 Files.Read.All, 목록 항목·페이지용 Sites.Read.All). 보통 상위 OAuthTokenResolver에서 연결하는데, 이 컴포넌트가 평문 문자열을 만들어내죠. Secret도 받아서 내부에서 해석해요.

오류 처리와 동시성

raise_on_failure(기본 True): False로 두면 실패한 항목은 로그만 남기고 건너뛰어서 나머지 항목은 계속 돌려받아요.
max_retries(기본 3): 스로틀(HTTP 429)과 일시적 서버 오류에서 재시도해요.
max_concurrent_requests(기본 5): run_async가 동시에 가져올 항목 수를 제한해서 Microsoft Graph 속도 제한을 피해요. 항목을 하나씩 가져오는 동기식 run에는 영향이 없어요.

설치

Microsoft SharePoint 통합을 설치해요.

pip install microsoft-sharepoint-haystack

더 알아보기 (Learn more)

단독으로 쓰기

아래 access_token은 사용자별 위임된 Microsoft Graph bearer 토큰이에요. 원시 web_url 문자열이나 MSSharePointRetriever가 만든 Document를 넘길 수 있어요.

from haystack_integrations.components.fetchers.microsoft_sharepoint import (
    MSSharePointFetcher,
)


fetcher = MSSharePointFetcher()

result = fetcher.run(
    access_token="my-delegated-graph-token",
    targets=[
        "https://contoso.sharepoint.com/sites/contoso-team/contoso-designs.docx",
    ],
)

for stream in result["streams"]:
    print(stream.meta["file_name"], stream.meta["content_type"])

파이프라인에서 쓰기

다음 검색 파이프라인은 통합 전체를 이어줘요. OAuthTokenResolver가 토큰을 만들고, MSSharePointRetriever가 SharePoint를 검색하며, MSSharePointFetcher가 일치하는 항목을 내려받고, FileTypeRouter가 각 ByteStream을 올바른 변환기로 보내죠. Resolver의 하나뿐인 access_token 출력이 Retriever와 Fetcher 양쪽에 공급되는 점을 눈여겨보세요.

from haystack import Pipeline
from haystack.utils import Secret
from haystack.components.routers import FileTypeRouter
from haystack.components.converters import PyPDFToDocument, DOCXToDocument

from haystack_integrations.components.connectors.oauth import OAuthTokenResolver
from haystack_integrations.utils.oauth import OAuthRefreshTokenSource
from haystack_integrations.components.retrievers.microsoft_sharepoint import (
    MSSharePointRetriever,
)
from haystack_integrations.components.fetchers.microsoft_sharepoint import (
    MSSharePointFetcher,
)


pipeline = Pipeline()
pipeline.add_component(
    "resolver",
    OAuthTokenResolver(
        token_source=OAuthRefreshTokenSource(
            token_url="https://login.microsoftonline.com/common/oauth2/v2.0/token",
            client_id="aaa-bbb-ccc",
            refresh_token=Secret.from_env_var("MS_REFRESH_TOKEN"),
            scopes=[
                "https://graph.microsoft.com/Files.Read.All",
                "https://graph.microsoft.com/Sites.Read.All",
                "offline_access",
            ],
        ),
    ),
)
pipeline.add_component("retriever", MSSharePointRetriever(top_k=5))
pipeline.add_component("fetcher", MSSharePointFetcher())
pipeline.add_component(
    "router",
    FileTypeRouter(
        mime_types=[
            "application/pdf",
            "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
        ],
    ),
)
pipeline.add_component("pdf_converter", PyPDFToDocument())
pipeline.add_component("docx_converter", DOCXToDocument())

# The same token feeds both the retriever and the fetcher.
pipeline.connect("resolver.access_token", "retriever.access_token")
pipeline.connect("resolver.access_token", "fetcher.access_token")

# The retrieved documents become the fetcher's targets.
pipeline.connect("retriever.documents", "fetcher.targets")

# Route each fetched ByteStream to the matching converter.
pipeline.connect("fetcher.streams", "router.sources")
pipeline.connect("router.application/pdf", "pdf_converter.sources")
pipeline.connect(
    "router.application/vnd.openxmlformats-officedocument.wordprocessingml.document",
    "docx_converter.sources",
)

result = pipeline.run({"retriever": {"query": "quarterly roadmap"}})