Source code for langchain.document_loaders.gcs_file
import os
import tempfile
from typing import Callable, List, Optional
from langchain.docstore.document import Document
from langchain.document_loaders.base import BaseLoader
from langchain.document_loaders.unstructured import UnstructuredFileLoader
[docs]class GCSFileLoader(BaseLoader):
    """Load from GCS file."""
[docs]    def __init__(
        self,
        project_name: str,
        bucket: str,
        blob: str,
        loader_func: Optional[Callable[[str], BaseLoader]] = None,
    ):
        """Initialize with bucket and key name.
        Args:
            project_name: The name of the project to load
            bucket: The name of the GCS bucket.
            blob: The name of the GCS blob to load.
            loader_func: A loader function that instatiates a loader based on a
                file_path argument. If nothing is provided, the
                UnstructuredFileLoader is used.
        Examples:
            To use an alternative PDF loader:
            >> from from langchain.document_loaders import PyPDFLoader
            >> loader = GCSFileLoader(..., loader_func=PyPDFLoader)
            To use UnstructuredFileLoader with additional arguments:
            >> loader = GCSFileLoader(...,
            >>      loader_func=lambda x: UnstructuredFileLoader(x, mode="elements"))
        """
        self.bucket = bucket
        self.blob = blob
        self.project_name = project_name
        def default_loader_func(file_path: str) -> BaseLoader:
            return UnstructuredFileLoader(file_path)
        self._loader_func = loader_func if loader_func else default_loader_func
[docs]    def load(self) -> List[Document]:
        """Load documents."""
        try: