Source code for langchain.retrievers.self_query.supabase
from typing import Any, Dict, Tuple
from langchain.chains.query_constructor.ir import (
    Comparator,
    Comparison,
    Operation,
    Operator,
    StructuredQuery,
    Visitor,
)
[docs]class SupabaseVectorTranslator(Visitor):
    """Translate Langchain filters to Supabase PostgREST filters."""
    allowed_operators = [Operator.AND, Operator.OR]
    """Subset of allowed logical operators."""
    allowed_comparators = [
        Comparator.EQ,
        Comparator.NE,
        Comparator.GT,
        Comparator.GTE,
        Comparator.LT,
        Comparator.LTE,
        Comparator.LIKE,
    ]
    """Subset of allowed logical comparators."""
    metadata_column = "metadata"
    def _map_comparator(self, comparator: Comparator) -> str:
        """
        Maps Langchain comparator to PostgREST comparator:
        https://postgrest.org/en/stable/references/api/tables_views.html#operators
        """
        postgrest_comparator = {
            Comparator.EQ: "eq",
            Comparator.NE: "neq",
            Comparator.GT: "gt",
            Comparator.GTE: "gte",
            Comparator.LT: "lt",
            Comparator.LTE: "lte",
            Comparator.LIKE: "like",
        }.get(comparator)
        if postgrest_comparator is None:
            raise Exception(
                f"Comparator '{comparator}' is not currently "
                "supported in Supabase Vector"
            )
        return postgrest_comparator
    def _get_json_operator(self, value: Any) -> str:
        if isinstance(value, str):
            return "->>"