Source code for langchain.vectorstores.starrocks
"""Wrapper around open source StarRocks VectorSearch capability."""
from __future__ import annotations
import json
import logging
from hashlib import sha1
from threading import Thread
from typing import Any, Dict, Iterable, List, Optional, Tuple
from pydantic import BaseSettings
from langchain.docstore.document import Document
from langchain.embeddings.base import Embeddings
from langchain.vectorstores.base import VectorStore
logger = logging.getLogger()
DEBUG = False
[docs]def has_mul_sub_str(s: str, *args: Any) -> bool:
    """
    Check if a string has multiple substrings.
    Args:
        s: The string to check
        *args: The substrings to check for in the string
    Returns:
        bool: True if all substrings are present in the string, False otherwise
    """
    for a in args:
        if a not in s:
            return False
    return True
[docs]def debug_output(s: Any) -> None:
    """
    Print a debug message if DEBUG is True.
    Args:
        s: The message to print
    """
    if DEBUG:
        print(s)
[docs]def get_named_result(connection: Any, query: str) -> List[dict[str, Any]]:
    """
    Get a named result from a query.
    Args:
        connection: The connection to the database
        query: The query to execute
    Returns:
        List[dict[str, Any]]: The result of the query
    """
    cursor = connection.cursor()
    cursor.execute(query)
    columns = cursor.description
    result = []
    for value in cursor.fetchall():
        r = {}
        for idx, datum in enumerate(value):