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

1"""Vendor-neutral metadata filter primitives for vector search.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from enum import StrEnum 

7from typing import Any 

8 

9from lexigram.contracts.data.types import LogicalOperator 

10 

11 

12class FilterOperator(StrEnum): 

13 """Comparison operators for metadata filtering.""" 

14 

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" 

25 

26 

27@dataclass(frozen=True, slots=True) 

28class MetadataCondition: 

29 """A single metadata filter condition.""" 

30 

31 field: str 

32 operator: FilterOperator 

33 value: Any 

34 

35 

36@dataclass(frozen=True, slots=True) 

37class MetadataConditionGroup: 

38 """A group of conditions combined with AND/OR.""" 

39 

40 logical_operator: LogicalOperator 

41 conditions: tuple[MetadataCondition | MetadataConditionGroup, ...] 

42 

43 

44# Union type for protocol signatures 

45MetadataFilter = MetadataCondition | MetadataConditionGroup 

46 

47 

48class Filter: 

49 """Static factory for building metadata filters. 

50 

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 """ 

63 

64 @staticmethod 

65 def eq(field_name: str, value: Any) -> MetadataCondition: 

66 """Equality: field == value.""" 

67 return MetadataCondition(field_name, FilterOperator.EQ, value) 

68 

69 @staticmethod 

70 def ne(field_name: str, value: Any) -> MetadataCondition: 

71 """Not equal: field != value.""" 

72 return MetadataCondition(field_name, FilterOperator.NE, value) 

73 

74 @staticmethod 

75 def gt(field_name: str, value: Any) -> MetadataCondition: 

76 """Greater than: field > value.""" 

77 return MetadataCondition(field_name, FilterOperator.GT, value) 

78 

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) 

83 

84 @staticmethod 

85 def lt(field_name: str, value: Any) -> MetadataCondition: 

86 """Less than: field < value.""" 

87 return MetadataCondition(field_name, FilterOperator.LT, value) 

88 

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) 

93 

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) 

98 

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) 

103 

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) 

108 

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) 

113 

114 @staticmethod 

115 def and_(*conditions: MetadataFilter) -> MetadataConditionGroup: 

116 """Logical AND of conditions.""" 

117 return MetadataConditionGroup(LogicalOperator.AND, tuple(conditions)) 

118 

119 @staticmethod 

120 def or_(*conditions: MetadataFilter) -> MetadataConditionGroup: 

121 """Logical OR of conditions.""" 

122 return MetadataConditionGroup(LogicalOperator.OR, tuple(conditions)) 

123 

124 

125__all__ = [ 

126 "Filter", 

127 "FilterOperator", 

128 "LogicalOperator", 

129 "MetadataCondition", 

130 "MetadataConditionGroup", 

131 "MetadataFilter", 

132]