Source code for langchain.vectorstores.singlestoredb
from __future__ import annotations
import json
from typing import (
    Any,
    Callable,
    ClassVar,
    Collection,
    Iterable,
    List,
    Optional,
    Tuple,
    Type,
)
from sqlalchemy.pool import QueuePool
from langchain.callbacks.manager import (
    AsyncCallbackManagerForRetrieverRun,
    CallbackManagerForRetrieverRun,
)
from langchain.docstore.document import Document
from langchain.schema.embeddings import Embeddings
from langchain.schema.vectorstore import VectorStore, VectorStoreRetriever
from langchain.vectorstores.utils import DistanceStrategy
DEFAULT_DISTANCE_STRATEGY = DistanceStrategy.DOT_PRODUCT
ORDERING_DIRECTIVE: dict = {
    DistanceStrategy.EUCLIDEAN_DISTANCE: "",
    DistanceStrategy.DOT_PRODUCT: "DESC",
}
[docs]class SingleStoreDB(VectorStore):
    """`SingleStore DB` vector store.
    The prerequisite for using this class is the installation of the ``singlestoredb``
    Python package.
    The SingleStoreDB vectorstore can be created by providing an embedding function and
    the relevant parameters for the database connection, connection pool, and
    optionally, the names of the table and the fields to use.
    """
    def _get_connection(self: SingleStoreDB) -> Any:
        try:
            import singlestoredb as s2
        except ImportError:
            raise ImportError(
                "Could not import singlestoredb python package. "
                "Please install it with `pip install singlestoredb`."
            )
        return s2.connect(**self.connection_kwargs)
[docs]    def __init__(
        self,
        embedding: Embeddings,
        *,