GoogleDriveFetcher

Drive API v3를 통해 Google Drive 파일의 전체 콘텐츠를 가져와 ByteStream으로 반환해요.

출처: 문서

본문

GoogleDriveFetcher는 Drive API v3를 통해 Google Drive 파일의 전체 콘텐츠를 다운로드하고 ByteStream 객체를 반환해요. 다운스트림 컨버터에 바로 쓸 수 있죠.

이것은 메타데이터(그리고 선택적으로 내보낸 텍스트)만 반환하는 GoogleDriveRetriever를 보완해요. 리트리버의 documents(또는 파일 id/Drive URL 리스트)를 페처에 연결하면 기본 콘텐츠를 다운로드해요. 페처는 각 파일의 mime 타입에 따라 분기해요:

바이너리 파일(PDF, DOCX, 이미지 등)은 files.get?alt=media로 그대로 다운로드해요.
네이티브 Google Docs/Sheets/Slides는 files.export로 내보내며, 기본적으로 Office 형식(DOCX/XLSX/PPTX)으로, export_mime_types로 구성 가능해요.
폴더 및 기타 다운로드 불가 Google 유형(Forms, Sites 등)은 건너뜁니다.

각 ByteStream의 meta에는 file_id, web_url, file_name, content_type이 담겨요. 출력이 혼합 유형의 ByteStream 리스트이므로, 일반적인 다음 단계는 각 스트림을 올바른 컨버터(PyPDFToDocument, DOCXToDocument, XLSXToDocument, 또는 PPTXToDocument)로 보내는 FileTypeRouter예요.

Authentication

페처는 run 입력으로 사용자별 access_token을 받아요. 토큰은 파일 콘텐츠 읽기를 허용하는 위임된 Google OAuth 범위(예: https://www.googleapis.com/auth/drive.readonly)를 가져야 해요. 일반적으로 업스트림 OAuthTokenResolver에서 연결하며, 일반 문자열을 내보내요. Secret도 허용되고 내부적으로 해결돼요.

Error handling and concurrency

raise_on_failure(기본 True): False면 실패한 가져오기는 로그되고 파일을 건너뛰므로 나머지 파일은 여전히 반환돼요.
max_retries(기본 3): 제한(HTTP 429) 및 일시적 서버 에러에서 재시도해요.
max_concurrent_requests(기본 5): run_async가 동시에 가져오는 파일 수를 제한해 Drive rate limit을 피해요. 파일을 하나씩 가져오는 동기 run에는 영향이 없어요.
export_mime_types: 기본 네이티브-Google-to-Office 내보내기 매핑을 재정의해요. Drive는 단일 내보내기를 10MB로 제한해요.

Installation

pip install google-drive-haystack

대표적인 파이프라인 위치: GoogleDriveRetriever 뒤, Router 또는 File Converter 앞
필수 init 변수: 없음
필수 run 변수: access_token(위임된 Google OAuth bearer 토큰) / targets(GoogleDriveRetriever에서 온 Document 또는 원본 Google Drive 파일 id/URL 리스트)
출력 변수: streams — 가져온 콘텐츠를 담은 ByteStream 리스트
API reference: Google Drive
패키지명: google-drive-haystack

Usage

On its own

아래 access_token은 사용자별 위임된 Google OAuth bearer 토큰이에요. 원본 파일 id/Drive URL 또는 GoogleDriveRetriever가 만든 Document를 전달할 수 있어요.

from haystack_integrations.components.fetchers.google_drive import GoogleDriveFetcher

fetcher = GoogleDriveFetcher()
result = fetcher.run(
    access_token="my-delegated-google-token",
    targets=[
        "https://drive.google.com/file/d/1AbCdEfGhIjKlMnOpQrStUvWxYz/view",
    ],
)
for stream in result["streams"]:
    print(stream.meta["file_name"], stream.meta["content_type"])

In a pipeline

다음 쿼리 파이프라인은 전체 통합을 하나로 묶어요: OAuthTokenResolver가 토큰을 제공하고, GoogleDriveRetriever가 Drive를 검색하고, GoogleDriveFetcher가 일치하는 파일을 다운로드하며, FileTypeRouter가 각 ByteStream을 올바른 컨버터로 보내요. 리졸버의 단일 access_token 출력이 리트리버와 페처 둘 다에 공급된다는 점에 주의하세요.

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.google_drive import (
    GoogleDriveRetriever,
)
from haystack_integrations.components.fetchers.google_drive import GoogleDriveFetcher

pipeline = Pipeline()
pipeline.add_component(
    "resolver",
    OAuthTokenResolver(
        token_source=OAuthRefreshTokenSource(
            token_url="https://oauth2.googleapis.com/token",
            client_id="aaa-bbb-ccc",
            refresh_token=Secret.from_env_var("GOOGLE_REFRESH_TOKEN"),
            scopes=["https://www.googleapis.com/auth/drive.readonly"],
        ),
    ),
)
pipeline.add_component("retriever", GoogleDriveRetriever(top_k=5))
pipeline.add_component("fetcher", GoogleDriveFetcher())
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"}})

GoogleDriveFetcher