Couchbase

CouchbaseReader

Bases: BaseReader

Couchbase文档加载器。

从Couchbase集群加载数据到LlamaIndex使用的文档。

Parameters:

Name	Type	Description	Default
client(Optional[Any])		用于连接的Couchbase客户端。 如果未提供，将根据connection_string和数据库凭据创建客户端。	required
connection_string	Optional[str]	连接到Couchbase集群的连接字符串。	None
db_username	Optional[str]	连接到Couchbase集群的用户名。	None
db_password	Optional[str]	连接到Couchbase集群的密码。	None

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

class CouchbaseReader(BaseReader):
    """Couchbase文档加载器。

    从Couchbase集群加载数据到LlamaIndex使用的文档。

    Args:
        client(Optional[Any]): 用于连接的Couchbase客户端。
            如果未提供，将根据connection_string和数据库凭据创建客户端。
        connection_string (Optional[str]): 连接到Couchbase集群的连接字符串。
        db_username (Optional[str]): 连接到Couchbase集群的用户名。
        db_password (Optional[str]): 连接到Couchbase集群的密码。"""

    def __init__(
        self,
        client: Optional[Any] = None,
        connection_string: Optional[str] = None,
        db_username: Optional[str] = None,
        db_password: Optional[str] = None,
    ) -> None:
        """初始化Couchbase文档加载程序。"""
        import_err_msg = "`couchbase` package not found, please run `pip install --upgrade couchbase`"
        try:
            from couchbase.auth import PasswordAuthenticator
            from couchbase.cluster import Cluster
            from couchbase.options import ClusterOptions
        except ImportError:
            raise ImportError(import_err_msg)

        if not client:
            if not connection_string or not db_username or not db_password:
                raise ValueError(
                    "You need to pass either a couchbase client or connection_string and credentials must be provided."
                )
            else:
                auth = PasswordAuthenticator(
                    db_username,
                    db_password,
                )

                self._client: Cluster = Cluster(connection_string, ClusterOptions(auth))
        else:
            self._client = client

    def lazy_load_data(
        self,
        query: str,
        text_fields: Optional[List[str]] = None,
        metadata_fields: Optional[List[str]] = [],
    ) -> Iterable[Document]:
        """从Couchbase集群中懒加载数据。

Args:
    query (str): 要执行的SQL++查询。
    text_fields (Optional[List[str]]): 要写入文档的`text`字段的列。默认情况下，将写入所有列。
    metadata_fields (Optional[List[str]]): 要写入文档的`metadata`字段的列。默认情况下，不会写入任何列。
"""
        from datetime import timedelta

        if not query:
            raise ValueError("Query must be provided.")

        # Ensure connection to Couchbase cluster
        self._client.wait_until_ready(timedelta(seconds=5))

        # Run SQL++ Query
        result = self._client.query(query)
        for row in result:
            if not text_fields:
                text_fields = list(row.keys())

            metadata = {field: row[field] for field in metadata_fields}

            document = "\n".join(
                f"{k}: {v}" for k, v in row.items() if k in text_fields
            )

            yield (Document(text=document, metadata=metadata))

    def load_data(
        self,
        query: str,
        text_fields: Optional[List[str]] = None,
        metadata_fields: Optional[List[str]] = None,
    ) -> List[Document]:
        """从Couchbase集群加载数据。

Args:
    query (str): 要执行的SQL++查询。
    text_fields (Optional[List[str]]): 要写入文档的`text`字段的列。默认情况下，将写入所有列。
    metadata_fields (Optional[List[str]]): 要写入文档的`metadata`字段的列。默认情况下，不写入任何列。
"""
        return list(self.lazy_load_data(query, text_fields, metadata_fields))

lazy_load_data

lazy_load_data(
    query: str,
    text_fields: Optional[List[str]] = None,
    metadata_fields: Optional[List[str]] = [],
) -> Iterable[Document]

从Couchbase集群中懒加载数据。

Parameters:

Name	Type	Description	Default
query	str	要执行的SQL++查询。	required
text_fields	Optional[List[str]]	要写入文档的text字段的列。默认情况下，将写入所有列。	None
metadata_fields	Optional[List[str]]	要写入文档的metadata字段的列。默认情况下，不会写入任何列。	[]

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

    def lazy_load_data(
        self,
        query: str,
        text_fields: Optional[List[str]] = None,
        metadata_fields: Optional[List[str]] = [],
    ) -> Iterable[Document]:
        """从Couchbase集群中懒加载数据。

Args:
    query (str): 要执行的SQL++查询。
    text_fields (Optional[List[str]]): 要写入文档的`text`字段的列。默认情况下，将写入所有列。
    metadata_fields (Optional[List[str]]): 要写入文档的`metadata`字段的列。默认情况下，不会写入任何列。
"""
        from datetime import timedelta

        if not query:
            raise ValueError("Query must be provided.")

        # Ensure connection to Couchbase cluster
        self._client.wait_until_ready(timedelta(seconds=5))

        # Run SQL++ Query
        result = self._client.query(query)
        for row in result:
            if not text_fields:
                text_fields = list(row.keys())

            metadata = {field: row[field] for field in metadata_fields}

            document = "\n".join(
                f"{k}: {v}" for k, v in row.items() if k in text_fields
            )

            yield (Document(text=document, metadata=metadata))

load_data

load_data(
    query: str,
    text_fields: Optional[List[str]] = None,
    metadata_fields: Optional[List[str]] = None,
) -> List[Document]

从Couchbase集群加载数据。

Parameters:

Name	Type	Description	Default
query	str	要执行的SQL++查询。	required
text_fields	Optional[List[str]]	要写入文档的text字段的列。默认情况下，将写入所有列。	None
metadata_fields	Optional[List[str]]	要写入文档的metadata字段的列。默认情况下，不写入任何列。	None

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

    def load_data(
        self,
        query: str,
        text_fields: Optional[List[str]] = None,
        metadata_fields: Optional[List[str]] = None,
    ) -> List[Document]:
        """从Couchbase集群加载数据。

Args:
    query (str): 要执行的SQL++查询。
    text_fields (Optional[List[str]]): 要写入文档的`text`字段的列。默认情况下，将写入所有列。
    metadata_fields (Optional[List[str]]): 要写入文档的`metadata`字段的列。默认情况下，不写入任何列。
"""
        return list(self.lazy_load_data(query, text_fields, metadata_fields))