[ Web Proxy ]
URL:
Viewing: https://raw.githubusercontent.com/feast-dev/feast/master/sdk/python/feast/filter_models.py [Back]  [Original]

from abc import ABC, abstractmethod
from typing import Any, Dict, List, Literal, Optional, Union

from pydantic import BaseModel


class ComparisonFilter(BaseModel):
    """A filter that compares a metadata field against a value.

    :param type: The comparison operator to apply
    :param key: The metadata field name to filter on
    :param value: The value to compare against
    """

    type: Literal["eq", "ne", "gt", "gte", "lt", "lte", "in", "nin"]
    key: str
    value: Any


class CompoundFilter(BaseModel):
    """A filter that combines multiple filters with a logical operator.

    :param type: The logical operator ("and" requires all filters match,
        "or" requires any filter matches)
    :param filters: The list of filters to combine
    """

    type: Literal["and", "or"]
    filters: List[Union[ComparisonFilter, "CompoundFilter"]]


CompoundFilter.model_rebuild()

FilterType = Optional[Union[ComparisonFilter, CompoundFilter]]


class FilterTranslator(ABC):
    """Abstract base for translating Feast filters into backend-native expressions.

    Each online store backend provides a concrete subclass that knows how to
    turn :class:`ComparisonFilter` / :class:`CompoundFilter` trees into the
    query language understood by the underlying database or search engine.

    The :meth:`translate` entry-point handles ``None`` via :meth:`_empty` and
    delegates to :meth:`translate_comparison` / :meth:`translate_compound`
    through the shared :meth:`_dispatch` recursion, so subclasses only need to
    implement the three abstract leaf methods.
    """

    @abstractmethod
    def translate(self, filters: FilterType) -> Any:
        """Return the backend-native filter expression."""
        ...

    def _dispatch(self, filter_obj: Union[ComparisonFilter, CompoundFilter]) -> Any:
        """Route to comparison or compound handler (used for recursion)."""
        if isinstance(filter_obj, ComparisonFilter):
            return self.translate_comparison(filter_obj)
        elif isinstance(filter_obj, CompoundFilter):
            return self.translate_compound(filter_obj)
        raise ValueError(f"Unknown filter type: {type(filter_obj)}")

    @abstractmethod
    def translate_comparison(self, f: ComparisonFilter) -> Any:
        """Translate a single comparison into a backend-native expression."""
        ...

    @abstractmethod
    def translate_compound(self, f: CompoundFilter) -> Any:
        """Translate a compound (AND/OR) filter into a backend-native expression."""
        ...


def filters_contain_numeric_comparison(
    filter_obj: Union[ComparisonFilter, CompoundFilter],
) -> bool:
    """Return True if any leaf comparison uses a numeric value."""
    if isinstance(filter_obj, ComparisonFilter):
        return isinstance(filter_obj.value, (int, float)) and not isinstance(
            filter_obj.value, bool
        )
    if isinstance(filter_obj, CompoundFilter):
        return any(filters_contain_numeric_comparison(f) for f in filter_obj.filters)
    return False


def convert_dict_to_filter(
    filter_dict: Dict[str, Any],
) -> Union[ComparisonFilter, CompoundFilter]:
    """Convert a raw dict (e.g. from OpenAI-compatible JSON) into a typed filter object."""
    filter_type = filter_dict.get("type")
    if filter_type in ("and", "or"):
        return CompoundFilter(
            type=filter_type,
            filters=[convert_dict_to_filter(f) for f in filter_dict["filters"]],
        )
    return ComparisonFilter(
        type=filter_dict["type"],
        key=filter_dict["key"],
        value=filter_dict["value"],
    )

Web Proxy Viewer  |  New URL  |  Original Page