Source code for langchain.vectorstores.starrocks
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 langchain.docstore.document import Document
from langchain.pydantic_v1 import BaseSettings
from langchain.schema.embeddings import Embeddings
from langchain.schema.vectorstore 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):
            k = columns[idx][0]