Source code for langchain_community.document_loaders.couchbase

import logging
from typing import Iterator, List, Optional

from langchain_core.documents import Document

from langchain_community.document_loaders.base import BaseLoader

logger = logging.getLogger(__name__)

[docs]class CouchbaseLoader(BaseLoader): """Load documents from `Couchbase`. Each document represents one row of the result. The `page_content_fields` are written into the `page_content`of the document. The `metadata_fields` are written into the `metadata` of the document. By default, all columns are written into the `page_content` and none into the `metadata`. """
[docs] def __init__( self, connection_string: str, db_username: str, db_password: str, query: str, *, page_content_fields: Optional[List[str]] = None, metadata_fields: Optional[List[str]] = None, ) -> None: """Initialize Couchbase document loader. Args: connection_string (str): The connection string to the Couchbase cluster. db_username (str): The username to connect to the Couchbase cluster. db_password (str): The password to connect to the Couchbase cluster. query (str): The SQL++ query to execute. page_content_fields (Optional[List[str]]): The columns to write into the `page_content` field of the document. By default, all columns are written. metadata_fields (Optional[List[str]]): The columns to write into the `metadata` field of the document. By default, no columns are written. """ try: from couchbase.auth import PasswordAuthenticator from couchbase.cluster import Cluster from couchbase.options import ClusterOptions except ImportError as e: raise ImportError( "Could not import couchbase package." "Please install couchbase SDK with `pip install couchbase`." ) from e if not connection_string: raise ValueError("connection_string must be provided.") if not db_username: raise ValueError("db_username must be provided.") if not db_password: raise ValueError("db_password must be provided.") auth = PasswordAuthenticator( db_username, db_password, ) self.cluster: Cluster = Cluster(connection_string, ClusterOptions(auth)) self.query = query self.page_content_fields = page_content_fields self.metadata_fields = metadata_fields
[docs] def lazy_load(self) -> Iterator[Document]: """Load Couchbase data into Document objects lazily.""" from datetime import timedelta # Ensure connection to Couchbase cluster self.cluster.wait_until_ready(timedelta(seconds=5)) # Run SQL++ Query result = self.cluster.query(self.query) for row in result: metadata_fields = self.metadata_fields page_content_fields = self.page_content_fields if not page_content_fields: page_content_fields = list(row.keys()) if not metadata_fields: metadata_fields = [] metadata = {field: row[field] for field in metadata_fields} document = "\n".join( f"{k}: {v}" for k, v in row.items() if k in page_content_fields ) yield (Document(page_content=document, metadata=metadata))