Source code for langchain.document_loaders.rocksetdb
from typing import Any, Callable, Iterator, List, Optional, Tuple
from langchain.document_loaders.base import BaseLoader
from langchain.schema import Document
[docs]def default_joiner(docs: List[Tuple[str, Any]]) -> str:
    """Default joiner for content columns."""
    return "\n".join([doc[1] for doc in docs])
[docs]class ColumnNotFoundError(Exception):
    """Column not found error."""
    def __init__(self, missing_key: str, query: str):
        super().__init__(f'Column "{missing_key}" not selected in query:\n{query}')
[docs]class RocksetLoader(BaseLoader):
    """Load from a `Rockset` database.
    To use, you should have the `rockset` python package installed.
    Example:
        .. code-block:: python
            # This code will load 3 records from the "langchain_demo"
            # collection as Documents, with the `text` column used as
            # the content
            from langchain.document_loaders import RocksetLoader
            from rockset import RocksetClient, Regions, models
            loader = RocksetLoader(
                RocksetClient(Regions.usw2a1, "<api key>"),
                models.QueryRequestSql(
                    query="select * from langchain_demo limit 3"
                ),
                ["text"]
            )
        )
    """
[docs]    def __init__(
        self,
        client: Any,
        query: Any,
        content_keys: List[str],
        metadata_keys: Optional[List[str]] = None,
        content_columns_joiner: Callable[[List[Tuple[str, Any]]], str] = default_joiner,
    ):