QueryTransformer Class — langchain Architecture

Source Code

libs/langchain/langchain_classic/chains/query_constructor/parser.py lines 78–254

class QueryTransformer(Transformer):
    """Transform a query string into an intermediate representation."""

    def __init__(
        self,
        *args: Any,
        allowed_comparators: Sequence[Comparator] | None = None,
        allowed_operators: Sequence[Operator] | None = None,
        allowed_attributes: Sequence[str] | None = None,
        **kwargs: Any,
    ):
        """Initialize the QueryTransformer.

        Args:
            *args: Positional arguments.
            allowed_comparators: Optional sequence of allowed comparators.
            allowed_operators: Optional sequence of allowed operators.
            allowed_attributes: Optional sequence of allowed attributes for comparators.
            **kwargs: Additional keyword arguments.
        """
        super().__init__(*args, **kwargs)
        self.allowed_comparators = allowed_comparators
        self.allowed_operators = allowed_operators
        self.allowed_attributes = allowed_attributes

    def program(self, *items: Any) -> tuple:
        """Transform the items into a tuple."""
        return items

    def func_call(self, func_name: Any, args: list) -> FilterDirective:
        """Transform a function name and args into a FilterDirective.

        Args:
            func_name: The name of the function.
            args: The arguments passed to the function.

        Returns:
            The filter directive.

        Raises:
            ValueError: If the function is a comparator and the first arg is not in the
            allowed attributes.
        """
        func = self._match_func_name(str(func_name))
        if isinstance(func, Comparator):
            if self.allowed_attributes and args[0] not in self.allowed_attributes:
                msg = (
                    f"Received invalid attributes {args[0]}. Allowed attributes are "
                    f"{self.allowed_attributes}"
                )
                raise ValueError(msg)
            return Comparison(comparator=func, attribute=args[0], value=args[1])
        if len(args) == 1 and func in (Operator.AND, Operator.OR):
            return args[0]
        return Operation(operator=func, arguments=args)

    def _match_func_name(self, func_name: str) -> Operator | Comparator:
        if func_name in set(Comparator):
            if (
                self.allowed_comparators is not None
                and func_name not in self.allowed_comparators
            ):
                msg = (
                    f"Received disallowed comparator {func_name}. Allowed "
                    f"comparators are {self.allowed_comparators}"
                )
                raise ValueError(msg)
            return Comparator(func_name)
        if func_name in set(Operator):
            if (
                self.allowed_operators is not None
                and func_name not in self.allowed_operators
            ):
                msg = (
                    f"Received disallowed operator {func_name}. Allowed operators"
                    f" are {self.allowed_operators}"
                )
                raise ValueError(msg)
            return Operator(func_name)
        msg = (
            f"Received unrecognized function {func_name}. Valid functions are "