-
Notifications
You must be signed in to change notification settings - Fork 1.4k
Expand file tree
/
Copy pathfilter_models.py
More file actions
101 lines (77 loc) · 3.52 KB
/
Copy pathfilter_models.py
File metadata and controls
101 lines (77 loc) · 3.52 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
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"],
)