Source code for langchain_core.structured_query

"""结构化查询语言的内部表示。"""
from __future__ import annotations

from abc import ABC, abstractmethod
from enum import Enum
from typing import Any, List, Optional, Sequence, Union

from langchain_core.pydantic_v1 import BaseModel


class Visitor(ABC):
    """使用访问者模式定义IR翻译的接口。"""

    allowed_comparators: Optional[Sequence[Comparator]] = None
    allowed_operators: Optional[Sequence[Operator]] = None

    def _validate_func(self, func: Union[Operator, Comparator]) -> None:
        if isinstance(func, Operator) and self.allowed_operators is not None:
            if func not in self.allowed_operators:
                raise ValueError(
                    f"Received disallowed operator {func}. Allowed "
                    f"comparators are {self.allowed_operators}"
                )
        if isinstance(func, Comparator) and self.allowed_comparators is not None:
            if func not in self.allowed_comparators:
                raise ValueError(
                    f"Received disallowed comparator {func}. Allowed "
                    f"comparators are {self.allowed_comparators}"
                )

    @abstractmethod
    def visit_operation(self, operation: Operation) -> Any:
        """操作。"""

    @abstractmethod
    def visit_comparison(self, comparison: Comparison) -> Any:
        """比较。"""

    @abstractmethod
    def visit_structured_query(self, structured_query: StructuredQuery) -> Any:
        """StructuredQuery。"""


def _to_snake_case(name: str) -> str:
    """将一个名称转换为蛇形命名法。"""
    snake_case = ""
    for i, char in enumerate(name):
        if char.isupper() and i != 0:
            snake_case += "_" + char.lower()
        else:
            snake_case += char.lower()
    return snake_case


class Expr(BaseModel):
    """所有表达式的基类。"""

    def accept(self, visitor: Visitor) -> Any:
        """接受一个访客。

参数：
    visitor: 要接受的访客

返回：
    访问结果
"""
        return getattr(visitor, f"visit_{_to_snake_case(self.__class__.__name__)}")(
            self
        )


class Operator(str, Enum):
    """操作的枚举器。"""

    AND = "and"
    OR = "or"
    NOT = "not"


class Comparator(str, Enum):
    """比较运算符的枚举器。"""

    EQ = "eq"
    NE = "ne"
    GT = "gt"
    GTE = "gte"
    LT = "lt"
    LTE = "lte"
    CONTAIN = "contain"
    LIKE = "like"
    IN = "in"
    NIN = "nin"


class FilterDirective(Expr, ABC):
    """过滤表达式。"""


class Comparison(FilterDirective):
    """与一个值进行比较。"""

    comparator: Comparator
    attribute: str
    value: Any

    def __init__(
        self, comparator: Comparator, attribute: str, value: Any, **kwargs: Any
    ) -> None:
        super().__init__(
            comparator=comparator, attribute=attribute, value=value, **kwargs
        )


class Operation(FilterDirective):
    """逻辑操作符用于其他指令。"""

    operator: Operator
    arguments: List[FilterDirective]

    def __init__(
        self, operator: Operator, arguments: List[FilterDirective], **kwargs: Any
    ):
        super().__init__(operator=operator, arguments=arguments, **kwargs)


class StructuredQuery(Expr):
    """结构化查询。"""

    query: str
    """查询字符串。"""
    filter: Optional[FilterDirective]
    """过滤表达式。"""
    limit: Optional[int]
    """结果数的限制。"""

    def __init__(
        self,
        query: str,
        filter: Optional[FilterDirective],
        limit: Optional[int] = None,
        **kwargs: Any,
    ):
        super().__init__(query=query, filter=filter, limit=limit, **kwargs)