Source code for langchain_community.document_loaders.obs_directory

# coding:utf-8
from typing import List, Optional

from langchain_core.documents import Document

from langchain_community.document_loaders.base import BaseLoader
from langchain_community.document_loaders.obs_file import OBSFileLoader


[docs]class OBSDirectoryLoader(BaseLoader):
    """Load from `Huawei OBS directory`."""

[docs]    def __init__(
        self,
        bucket: str,
        endpoint: str,
        config: Optional[dict] = None,
        prefix: str = "",
    ):
        """Initialize the OBSDirectoryLoader with the specified settings.

        Args:
            bucket (str): The name of the OBS bucket to be used.
            endpoint (str): The endpoint URL of your OBS bucket.
            config (dict): The parameters for connecting to OBS, provided as a dictionary. The dictionary could have the following keys:
                - "ak" (str, optional): Your OBS access key (required if `get_token_from_ecs` is False and bucket policy is not public read).
                - "sk" (str, optional): Your OBS secret key (required if `get_token_from_ecs` is False and bucket policy is not public read).
                - "token" (str, optional): Your security token (required if using temporary credentials).
                - "get_token_from_ecs" (bool, optional): Whether to retrieve the security token from ECS. Defaults to False if not provided. If set to True, `ak`, `sk`, and `token` will be ignored.
            prefix (str, optional): The prefix to be added to the OBS key. Defaults to "".

        Note:
            Before using this class, make sure you have registered with OBS and have the necessary credentials. The `ak`, `sk`, and `endpoint` values are mandatory unless `get_token_from_ecs` is True or the bucket policy is public read. `token` is required when using temporary credentials.
        Example:
            To create a new OBSDirectoryLoader:
            ```
            config = {
                "ak": "your-access-key",
                "sk": "your-secret-key"
            }
            ```
            directory_loader = OBSDirectoryLoader("your-bucket-name", "your-end-endpoint", config, "your-prefix")
        """  # noqa: E501
        try:
            from obs import ObsClient
        except ImportError:
            raise ImportError(
                "Could not import esdk-obs-python python package. "
                "Please install it with `pip install esdk-obs-python`."
            )
        if not config:
            config = dict()
        if config.get("get_token_from_ecs"):
            self.client = ObsClient(server=endpoint, security_provider_policy="ECS")
        else:
            self.client = ObsClient(
                access_key_id=config.get("ak"),
                secret_access_key=config.get("sk"),
                security_token=config.get("token"),
                server=endpoint,
            )

        self.bucket = bucket
        self.prefix = prefix

[docs]    def load(self) -> List[Document]:
        """Load documents."""
        max_num = 1000
        mark = None
        docs = []
        while True:
            resp = self.client.listObjects(
                self.bucket, prefix=self.prefix, marker=mark, max_keys=max_num
            )
            if resp.status < 300:
                for content in resp.body.contents:
                    loader = OBSFileLoader(self.bucket, content.key, client=self.client)
                    docs.extend(loader.load())
                if resp.body.is_truncated is True:
                    mark = resp.body.next_marker
                else:
                    break
        return docs