Source code for langchain_community.document_loaders.baiducloud_bos_file

import logging
import os
import tempfile
from typing import Any, Iterator

from langchain_core.documents import Document

from langchain_community.document_loaders.base import BaseLoader
from langchain_community.document_loaders.unstructured import UnstructuredFileLoader

logger = logging.getLogger(__name__)


[docs] class BaiduBOSFileLoader(BaseLoader): """Load from `Baidu Cloud BOS` file."""
[docs] def __init__(self, conf: Any, bucket: str, key: str): """Initialize with BOS config, bucket and key name. :param conf(BceClientConfiguration): BOS config. :param bucket(str): BOS bucket. :param key(str): BOS file key. """ self.conf = conf self.bucket = bucket self.key = key
[docs] def lazy_load(self) -> Iterator[Document]: """Load documents.""" try: from baidubce.services.bos.bos_client import BosClient except ImportError: raise ImportError( "Please using `pip install bce-python-sdk`" + " before import bos related package." ) # Initialize BOS Client client = BosClient(self.conf) with tempfile.TemporaryDirectory() as temp_dir: file_path = f"{temp_dir}/{self.bucket}/{self.key}" os.makedirs(os.path.dirname(file_path), exist_ok=True) # Download the file to a destination logger.debug(f"get object key {self.key} to file {file_path}") client.get_object_to_file(self.bucket, self.key, file_path) try: loader = UnstructuredFileLoader(file_path) documents = loader.load() return iter(documents) except Exception as ex: logger.error(f"load document error = {ex}") return iter([Document(page_content="")])