Source code for langchain_community.vectorstores.yellowbrick
from __future__ import annotations
import json
import logging
import uuid
import warnings
from itertools import repeat
from typing import (
    Any,
    Iterable,
    List,
    Optional,
    Tuple,
    Type,
)
from langchain_core.embeddings import Embeddings
from langchain_core.vectorstores import VectorStore
from langchain_community.docstore.document import Document
logger = logging.getLogger(__name__)
[docs]class Yellowbrick(VectorStore):
    """Wrapper around Yellowbrick as a vector database.
    Example:
        .. code-block:: python
            from langchain_community.vectorstores import Yellowbrick
            from langchain_community.embeddings.openai import OpenAIEmbeddings
            ...
    """
[docs]    def __init__(
        self,
        embedding: Embeddings,
        connection_string: str,
        table: str,
    ) -> None:
        """Initialize with yellowbrick client.
        Args:
            embedding: Embedding operator
            connection_string: Format 'postgres://username:password@host:port/database'
            table: Table used to store / retrieve embeddings from
        """
        import psycopg2
        if not isinstance(embedding, Embeddings):
            warnings.warn("embeddings input must be Embeddings object.")
        self.connection_string = connection_string
        self._table = table
        self._embedding = embedding
        self._connection = psycopg2.connect(connection_string)
        self.__post_init__()
    def __post_init__(
        self,
    ) -> None:
        """Initialize the store."""
        self.check_database_utf8()
        self.create_table_if_not_exists()
    def __del__(self) -> None:
        if self._connection:
            self._connection.close()