Coverage for src / lexigram / contracts / data / graph / filters.py: 80%

1"""Property filter primitives for graph nodes and edges.""" 

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 PropertyOperator(StrEnum): 

13 """Comparison operators for property 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 STARTS_WITH = "starts_with" 

26 ENDS_WITH = "ends_with" 

27 REGEX = "regex" 

28 

29 

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

31class PropertyCondition: 

32 """A single property filter condition.""" 

33 

34 field: str 

35 operator: PropertyOperator 

36 value: Any 

37 

38 

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

40class PropertyConditionGroup: 

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

42 

43 logical_operator: LogicalOperator 

44 conditions: tuple[PropertyCondition | PropertyConditionGroup, ...] 

45 

46 

47PropertyFilter = PropertyCondition | PropertyConditionGroup 

48 

49 

50class Prop: 

51 """Static factory for building property filters. 

52 

53 Usage: 

54 ```python 

55 f = Prop.and_( 

56 Prop.eq("active", True), 

57 Prop.gte("age", 21), 

58 Prop.or_( 

59 Prop.eq("role", "admin"), 

60 Prop.eq("role", "moderator"), 

61 ), 

62 ) 

63 ``` 

64 """ 

65 

66 @staticmethod 

67 def eq(field: str, value: Any) -> PropertyCondition: 

68 """Equality: field == value.""" 

69 return PropertyCondition(field, PropertyOperator.EQ, value) 

70 

71 @staticmethod 

72 def ne(field: str, value: Any) -> PropertyCondition: 

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

74 return PropertyCondition(field, PropertyOperator.NE, value) 

75 

76 @staticmethod 

77 def gt(field: str, value: Any) -> PropertyCondition: 

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

79 return PropertyCondition(field, PropertyOperator.GT, value) 

80 

81 @staticmethod 

82 def gte(field: str, value: Any) -> PropertyCondition: 

83 """Greater than or equal: field >= value.""" 

84 return PropertyCondition(field, PropertyOperator.GTE, value) 

85 

86 @staticmethod 

87 def lt(field: str, value: Any) -> PropertyCondition: 

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

89 return PropertyCondition(field, PropertyOperator.LT, value) 

90 

91 @staticmethod 

92 def lte(field: str, value: Any) -> PropertyCondition: 

93 """Less than or equal: field <= value.""" 

94 return PropertyCondition(field, PropertyOperator.LTE, value) 

95 

96 @staticmethod 

97 def in_(field: str, values: list[Any]) -> PropertyCondition: 

98 """Membership: field in values.""" 

99 return PropertyCondition(field, PropertyOperator.IN, values) 

100 

101 @staticmethod 

102 def not_in(field: str, values: list[Any]) -> PropertyCondition: 

103 """Exclusion: field not in values.""" 

104 return PropertyCondition(field, PropertyOperator.NOT_IN, values) 

105 

106 @staticmethod 

107 def exists(field: str, should_exist: bool = True) -> PropertyCondition: 

108 """Field existence check.""" 

109 return PropertyCondition(field, PropertyOperator.EXISTS, should_exist) 

110 

111 @staticmethod 

112 def contains(field: str, value: str) -> PropertyCondition: 

113 """String containment: value in field.""" 

114 return PropertyCondition(field, PropertyOperator.CONTAINS, value) 

115 

116 @staticmethod 

117 def starts_with(field: str, value: str) -> PropertyCondition: 

118 """String prefix match.""" 

119 return PropertyCondition(field, PropertyOperator.STARTS_WITH, value) 

120 

121 @staticmethod 

122 def ends_with(field: str, value: str) -> PropertyCondition: 

123 """String suffix match.""" 

124 return PropertyCondition(field, PropertyOperator.ENDS_WITH, value) 

125 

126 @staticmethod 

127 def regex(field: str, pattern: str) -> PropertyCondition: 

128 """Regular expression match.""" 

129 return PropertyCondition(field, PropertyOperator.REGEX, pattern) 

130 

131 @staticmethod 

132 def and_(*conditions: PropertyFilter) -> PropertyConditionGroup: 

133 """Logical AND of conditions.""" 

134 return PropertyConditionGroup(LogicalOperator.AND, tuple(conditions)) 

135 

136 @staticmethod 

137 def or_(*conditions: PropertyFilter) -> PropertyConditionGroup: 

138 """Logical OR of conditions.""" 

139 return PropertyConditionGroup(LogicalOperator.OR, tuple(conditions)) 

140 

141 

142__all__ = [ 

143 "Prop", 

144 "PropertyCondition", 

145 "PropertyConditionGroup", 

146 "PropertyFilter", 

147 "PropertyOperator", 

148]