Arango db

SimpleArangoDBReader

Bases: BaseReader

简单的arangodb读取器。

将每个ArangoDB文档连接成LlamaIndex使用的文档。

Args:

# host: (Union[str, List[str]]) 连接到数据库的url列表或url
# client: (Any) ArangoDB客户端

Source code in llama_index/readers/arango_db/base.py

class SimpleArangoDBReader(BaseReader):
    """简单的arangodb读取器。
    # 将每个ArangoDB文档连接成LlamaIndex使用的文档。

    # Args:
        # host: (Union[str, List[str]]) 连接到数据库的url列表或url
        # client: (Any) ArangoDB客户端"""

    def __init__(
        self, host: Optional[Union[str, List[str]]] = None, client: Optional[Any] = None
    ) -> None:
        """使用参数进行初始化。"""
        try:
            from arango import ArangoClient
        except ImportError as err:
            raise ImportError(
                "`arango` package not found, please run `pip install python-arango`"
            ) from err

        host = host or "http://127.0.0.1:8529"
        self.client = client or ArangoClient(hosts=host)
        self.client = cast(ArangoClient, self.client)

    def _flatten(self, texts: List[Union[str, List[str]]]) -> List[str]:
        result = []
        for text in texts:
            result += text if isinstance(text, list) else [text]
        return result

    def lazy_load(
        self,
        username: str,
        password: str,
        db_name: str,
        collection_name: str,
        field_names: List[str] = ["text"],
        separator: str = " ",
        query_dict: Optional[Dict] = {},
        max_docs: int = None,
        metadata_names: Optional[List[str]] = None,
    ) -> Iterator[Document]:
        """从ArangoDB中延迟加载数据。

Args:
    username (str)：用于凭据的用户名。
    password (str)：用于凭据的密码。
    db_name (str)：数据库的名称。
    collection_name (str)：集合的名称。
    field_names(List[str])：要连接的字段的名称。
        默认为["text"]
    separator (str)：字段之间要使用的分隔符。
        默认为" "
    query_dict (Optional[Dict])：用于过滤文档的查询。更多信息请阅读
    [docs](https://docs.python-arango.com/en/main/specs.html#arango.collection.StandardCollection.find)
        默认为空字典
    max_docs (int)：要加载的最大文档数量。
        默认为None（无限制）
    metadata_names (Optional[List[str]])：要添加到文档的metadata属性中的字段名称。
        默认为None
Returns:
    List[Document]：文档的列表。
"""
        db = self.client.db(name=db_name, username=username, password=password)
        collection = db.collection(collection_name)
        cursor = collection.find(filters=query_dict, limit=max_docs)
        for item in cursor:
            try:
                texts = [str(item[name]) for name in field_names]
            except KeyError as err:
                raise ValueError(
                    f"{err.args[0]} field not found in arangodb document."
                ) from err
            texts = self._flatten(texts)
            text = separator.join(texts)

            if metadata_names is None:
                yield Document(text=text)
            else:
                try:
                    metadata = {name: item[name] for name in metadata_names}
                except KeyError as err:
                    raise ValueError(
                        f"{err.args[0]} field not found in arangodb document."
                    ) from err
                yield Document(text=text, metadata=metadata)

    def load_data(
        self,
        username: str,
        password: str,
        db_name: str,
        collection_name: str,
        field_names: List[str] = ["text"],
        separator: str = " ",
        query_dict: Optional[Dict] = {},
        max_docs: int = None,
        metadata_names: Optional[List[str]] = None,
    ) -> List[Document]:
        """从ArangoDB加载数据。

Args:
    username (str): 用户名，用于凭据。
    password (str): 密码，用于凭据。
    db_name (str): 数据库的名称。
    collection_name (str): 集合的名称。
    field_names(List[str]): 要连接的字段的名称。
        默认为["text"]
    separator (str): 字段之间要使用的分隔符。
        默认为空字符串""
    query_dict (Optional[Dict]): 用于过滤文档的查询。更多信息请参阅[文档](https://docs.python-arango.com/en/main/specs.html#arango.collection.StandardCollection.find)
        默认为空字典
    max_docs (int): 要加载的最大文档数。
        默认为0（无限制）
    metadata_names (Optional[List[str]]): 要添加到文档的metadata属性中的字段名称。
        默认为None
Returns:
    List[Document]: 文档的列表。
"""
        return list(
            self.lazy_load(
                username,
                password,
                db_name,
                collection_name,
                field_names,
                separator,
                query_dict,
                max_docs,
                metadata_names,
            )
        )

lazy_load

lazy_load(
    username: str,
    password: str,
    db_name: str,
    collection_name: str,
    field_names: List[str] = ["text"],
    separator: str = " ",
    query_dict: Optional[Dict] = {},
    max_docs: int = None,
    metadata_names: Optional[List[str]] = None,
) -> Iterator[Document]

从ArangoDB中延迟加载数据。

Parameters:

Name	Type	Description	Default
[docs](https		//docs.python-arango.com/en/main/specs.html#arango.collection.StandardCollection.find) 默认为空字典	required

Returns: List[Document]：文档的列表。

Source code in llama_index/readers/arango_db/base.py

    def lazy_load(
        self,
        username: str,
        password: str,
        db_name: str,
        collection_name: str,
        field_names: List[str] = ["text"],
        separator: str = " ",
        query_dict: Optional[Dict] = {},
        max_docs: int = None,
        metadata_names: Optional[List[str]] = None,
    ) -> Iterator[Document]:
        """从ArangoDB中延迟加载数据。

Args:
    username (str)：用于凭据的用户名。
    password (str)：用于凭据的密码。
    db_name (str)：数据库的名称。
    collection_name (str)：集合的名称。
    field_names(List[str])：要连接的字段的名称。
        默认为["text"]
    separator (str)：字段之间要使用的分隔符。
        默认为" "
    query_dict (Optional[Dict])：用于过滤文档的查询。更多信息请阅读
    [docs](https://docs.python-arango.com/en/main/specs.html#arango.collection.StandardCollection.find)
        默认为空字典
    max_docs (int)：要加载的最大文档数量。
        默认为None（无限制）
    metadata_names (Optional[List[str]])：要添加到文档的metadata属性中的字段名称。
        默认为None
Returns:
    List[Document]：文档的列表。
"""
        db = self.client.db(name=db_name, username=username, password=password)
        collection = db.collection(collection_name)
        cursor = collection.find(filters=query_dict, limit=max_docs)
        for item in cursor:
            try:
                texts = [str(item[name]) for name in field_names]
            except KeyError as err:
                raise ValueError(
                    f"{err.args[0]} field not found in arangodb document."
                ) from err
            texts = self._flatten(texts)
            text = separator.join(texts)

            if metadata_names is None:
                yield Document(text=text)
            else:
                try:
                    metadata = {name: item[name] for name in metadata_names}
                except KeyError as err:
                    raise ValueError(
                        f"{err.args[0]} field not found in arangodb document."
                    ) from err
                yield Document(text=text, metadata=metadata)

load_data

load_data(
    username: str,
    password: str,
    db_name: str,
    collection_name: str,
    field_names: List[str] = ["text"],
    separator: str = " ",
    query_dict: Optional[Dict] = {},
    max_docs: int = None,
    metadata_names: Optional[List[str]] = None,
) -> List[Document]

从ArangoDB加载数据。

Parameters:

Name	Type	Description	Default
username	str	用户名，用于凭据。	required
password	str	密码，用于凭据。	required
db_name	str	数据库的名称。	required
collection_name	str	集合的名称。	required
field_names(List[str])		要连接的字段的名称。 默认为["text"]	required
separator	str	字段之间要使用的分隔符。 默认为空字符串""	' '
query_dict	Optional[Dict]	用于过滤文档的查询。更多信息请参阅文档 默认为空字典	{}
max_docs	int	要加载的最大文档数。 默认为0（无限制）	None
metadata_names	Optional[List[str]]	要添加到文档的metadata属性中的字段名称。 默认为None	None

Returns: List[Document]: 文档的列表。

Source code in llama_index/readers/arango_db/base.py

    def load_data(
        self,
        username: str,
        password: str,
        db_name: str,
        collection_name: str,
        field_names: List[str] = ["text"],
        separator: str = " ",
        query_dict: Optional[Dict] = {},
        max_docs: int = None,
        metadata_names: Optional[List[str]] = None,
    ) -> List[Document]:
        """从ArangoDB加载数据。

Args:
    username (str): 用户名，用于凭据。
    password (str): 密码，用于凭据。
    db_name (str): 数据库的名称。
    collection_name (str): 集合的名称。
    field_names(List[str]): 要连接的字段的名称。
        默认为["text"]
    separator (str): 字段之间要使用的分隔符。
        默认为空字符串""
    query_dict (Optional[Dict]): 用于过滤文档的查询。更多信息请参阅[文档](https://docs.python-arango.com/en/main/specs.html#arango.collection.StandardCollection.find)
        默认为空字典
    max_docs (int): 要加载的最大文档数。
        默认为0（无限制）
    metadata_names (Optional[List[str]]): 要添加到文档的metadata属性中的字段名称。
        默认为None
Returns:
    List[Document]: 文档的列表。
"""
        return list(
            self.lazy_load(
                username,
                password,
                db_name,
                collection_name,
                field_names,
                separator,
                query_dict,
                max_docs,
                metadata_names,
            )
        )