Source code for langchain_community.document_loaders.baiducloud_bos_directory

from typing import Any, Iterator

from langchain_core.documents import Document

from langchain_community.document_loaders.base import BaseLoader

[docs]class BaiduBOSDirectoryLoader(BaseLoader): """Load from `Baidu BOS directory`."""
[docs] def __init__(self, conf: Any, bucket: str, prefix: str = ""): """Initialize with BOS config, bucket and prefix. :param conf(BosConfig): BOS config. :param bucket(str): BOS bucket. :param prefix(str): prefix. """ self.conf = conf self.bucket = bucket self.prefix = prefix
[docs] def lazy_load(self) -> Iterator[Document]: """Load documents.""" try: from import BosClient except ImportError: raise ImportError( "Please install bce-python-sdk with `pip install bce-python-sdk`." ) client = BosClient(self.conf) contents = [] marker = "" while True: response = client.list_objects( bucket_name=self.bucket, prefix=self.prefix, marker=marker, max_keys=1000, ) contents_len = len(response.contents) contents.extend(response.contents) if response.is_truncated or contents_len < int(str(response.max_keys)): break marker = response.next_marker from langchain_community.document_loaders.baiducloud_bos_file import ( BaiduBOSFileLoader, ) for content in contents: if str(content.key).endswith("/"): continue loader = BaiduBOSFileLoader(self.conf, self.bucket, str(content.key)) yield loader.load()[0]