Coverage for src / lexigram / contracts / data / vector / filters.py: 81%
64 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-19 05:41 +0800
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-19 05:41 +0800
1"""Vendor-neutral metadata filter primitives for vector search."""
3from __future__ import annotations
5from dataclasses import dataclass
6from enum import StrEnum
7from typing import Any
9from lexigram.contracts.data.types import LogicalOperator
12class FilterOperator(StrEnum):
13 """Comparison operators for metadata filtering."""
15 EQ = "eq"
16 NE = "ne"
17 GT = "gt"
18 GTE = "gte"
19 LT = "lt"
20 LTE = "lte"
21 IN = "in"
22 NOT_IN = "not_in"
23 EXISTS = "exists"
24 CONTAINS = "contains"
27@dataclass(frozen=True, slots=True)
28class MetadataCondition:
29 """A single metadata filter condition."""
31 field: str
32 operator: FilterOperator
33 value: Any
36@dataclass(frozen=True, slots=True)
37class MetadataConditionGroup:
38 """A group of conditions combined with AND/OR."""
40 logical_operator: LogicalOperator
41 conditions: tuple[MetadataCondition | MetadataConditionGroup, ...]
44# Union type for protocol signatures
45MetadataFilter = MetadataCondition | MetadataConditionGroup
48class Filter:
49 """Static factory for building metadata filters.
51 Usage:
52 ```python
53 f = Filter.and_(
54 Filter.eq("category", "science"),
55 Filter.gte("year", 2020),
56 Filter.or_(
57 Filter.eq("status", "published"),
58 Filter.eq("status", "preprint"),
59 ),
60 )
61 ```
62 """
64 @staticmethod
65 def eq(field_name: str, value: Any) -> MetadataCondition:
66 """Equality: field == value."""
67 return MetadataCondition(field_name, FilterOperator.EQ, value)
69 @staticmethod
70 def ne(field_name: str, value: Any) -> MetadataCondition:
71 """Not equal: field != value."""
72 return MetadataCondition(field_name, FilterOperator.NE, value)
74 @staticmethod
75 def gt(field_name: str, value: Any) -> MetadataCondition:
76 """Greater than: field > value."""
77 return MetadataCondition(field_name, FilterOperator.GT, value)
79 @staticmethod
80 def gte(field_name: str, value: Any) -> MetadataCondition:
81 """Greater than or equal: field >= value."""
82 return MetadataCondition(field_name, FilterOperator.GTE, value)
84 @staticmethod
85 def lt(field_name: str, value: Any) -> MetadataCondition:
86 """Less than: field < value."""
87 return MetadataCondition(field_name, FilterOperator.LT, value)
89 @staticmethod
90 def lte(field_name: str, value: Any) -> MetadataCondition:
91 """Less than or equal: field <= value."""
92 return MetadataCondition(field_name, FilterOperator.LTE, value)
94 @staticmethod
95 def in_(field_name: str, values: list[Any]) -> MetadataCondition:
96 """Membership: field in values."""
97 return MetadataCondition(field_name, FilterOperator.IN, values)
99 @staticmethod
100 def not_in(field_name: str, values: list[Any]) -> MetadataCondition:
101 """Exclusion: field not in values."""
102 return MetadataCondition(field_name, FilterOperator.NOT_IN, values)
104 @staticmethod
105 def exists(field_name: str, should_exist: bool = True) -> MetadataCondition:
106 """Field existence: field exists (or not)."""
107 return MetadataCondition(field_name, FilterOperator.EXISTS, should_exist)
109 @staticmethod
110 def contains(field_name: str, value: str) -> MetadataCondition:
111 """String containment: value in field."""
112 return MetadataCondition(field_name, FilterOperator.CONTAINS, value)
114 @staticmethod
115 def and_(*conditions: MetadataFilter) -> MetadataConditionGroup:
116 """Logical AND of conditions."""
117 return MetadataConditionGroup(LogicalOperator.AND, tuple(conditions))
119 @staticmethod
120 def or_(*conditions: MetadataFilter) -> MetadataConditionGroup:
121 """Logical OR of conditions."""
122 return MetadataConditionGroup(LogicalOperator.OR, tuple(conditions))
125__all__ = [
126 "Filter",
127 "FilterOperator",
128 "LogicalOperator",
129 "MetadataCondition",
130 "MetadataConditionGroup",
131 "MetadataFilter",
132]