Coverage for src / lexigram / contracts / data / protocols.py: 0%

1"""Data access protocol class definitions.""" 

2 

3from __future__ import annotations 

4 

5from typing import ( 

6 TYPE_CHECKING, 

7 Any, 

8 Generic, 

9 Protocol, 

10 TypeVar, 

11 runtime_checkable, 

12) 

13 

14from lexigram.contracts.data.types import ( 

15 AndExpr as AndExpr, 

16) 

17from lexigram.contracts.data.types import ( 

18 CursorPaginationSpec as CursorPaginationSpec, 

19) 

20from lexigram.contracts.data.types import ( 

21 FieldContains as FieldContains, 

22) 

23from lexigram.contracts.data.types import ( 

24 FieldEq as FieldEq, 

25) 

26from lexigram.contracts.data.types import ( 

27 FieldGt as FieldGt, 

28) 

29from lexigram.contracts.data.types import ( 

30 FieldGte as FieldGte, 

31) 

32from lexigram.contracts.data.types import ( 

33 FieldIn as FieldIn, 

34) 

35from lexigram.contracts.data.types import ( 

36 FieldLt as FieldLt, 

37) 

38from lexigram.contracts.data.types import ( 

39 FieldLte as FieldLte, 

40) 

41from lexigram.contracts.data.types import ( 

42 FieldNeq as FieldNeq, 

43) 

44from lexigram.contracts.data.types import ( 

45 FilterExpression as FilterExpression, 

46) 

47from lexigram.contracts.data.types import ( 

48 NotExpr as NotExpr, 

49) 

50from lexigram.contracts.data.types import ( 

51 OrExpr as OrExpr, 

52) 

53from lexigram.contracts.data.types import ( 

54 PaginationSpec as PaginationSpec, 

55) 

56from lexigram.contracts.data.types import ( 

57 ProjectionSpec as ProjectionSpec, 

58) 

59from lexigram.contracts.data.types import ( 

60 SortSpecification as SortSpecification, 

61) 

62 

63if TYPE_CHECKING: 

64 from collections.abc import Awaitable 

65 

66_T_co = TypeVar("_T_co", covariant=True) 

67 

68 

69@runtime_checkable 

70class QueryFilterProtocol(Protocol): 

71 """Simple query filter for data access. 

72 

73 This is a simple field-value filter, not the full DDD SpecificationProtocol pattern. 

74 For composable specifications, use :class:`lexigram.contracts.domain.SpecificationProtocol`. 

75 """ 

76 

77 def __init__(self, field: str, value: Any) -> None: 

78 """Initialize filter. 

79 

80 Args: 

81 field: Field name to filter on 

82 value: Value to filter by 

83 """ 

84 ... 

85 

86 

87# --------------------------------------------------------------------------- 

88# Extension protocols (for data/ implementations) 

89# --------------------------------------------------------------------------- 

90 

91 

92@runtime_checkable 

93class FilterCompilerProtocol(Protocol, Generic[_T_co]): 

94 """Compiles a ``FilterExpression`` into a backend-specific query predicate. 

95 

96 Type parameter ``T`` is the compiled representation — a Python callable 

97 for in-memory repositories, an SQL fragment for SQL backends, etc. 

98 """ 

99 

100 def compile(self, expression: FilterExpression) -> _T_co: 

101 """Compile *expression* into a backend query predicate. 

102 

103 Args: 

104 expression: Filter expression to compile. 

105 

106 Returns: 

107 A backend-specific predicate or query fragment. 

108 """ 

109 ... 

110 

111 

112@runtime_checkable 

113class CursorCodecProtocol(Protocol): 

114 """Encodes and decodes opaque pagination cursor strings.""" 

115 

116 def encode(self, values: dict[str, Any]) -> str: 

117 """Encode *values* into an opaque cursor string. 

118 

119 Args: 

120 values: Mapping of field names to their cursor values. 

121 

122 Returns: 

123 An opaque cursor string. 

124 """ 

125 ... 

126 

127 def decode(self, cursor: str) -> dict[str, Any]: 

128 """Decode an opaque cursor string back to field values. 

129 

130 Args: 

131 cursor: An opaque cursor string produced by :meth:`encode`. 

132 

133 Returns: 

134 The original field-value mapping. 

135 """ 

136 ... 

137 

138 

139@runtime_checkable 

140class PaginatorProtocol(Protocol, Generic[_T_co]): 

141 """Performs a single page of cursor-based pagination.""" 

142 

143 def paginate( 

144 self, 

145 *, 

146 cursor: str | None = None, 

147 size: int = 20, 

148 **kwargs: Any, 

149 ) -> Awaitable[Any]: 

150 """Fetch one page of results. 

151 

152 Args: 

153 cursor: Opaque cursor from a previous page, or ``None`` for first. 

154 size: Number of results per page. 

155 **kwargs: Extra arguments forwarded to the fetch function. 

156 

157 Returns: 

158 An awaitable resolving to a ``CursorPage[T]``. 

159 """ 

160 ... 

161 

162 

163__all__ = [ 

164 "AndExpr", 

165 "CursorCodecProtocol", 

166 "CursorPaginationSpec", 

167 "FieldContains", 

168 "FieldEq", 

169 "FieldGt", 

170 "FieldGte", 

171 "FieldIn", 

172 "FieldLt", 

173 "FieldLte", 

174 "FieldNeq", 

175 "FilterCompilerProtocol", 

176 "FilterExpression", 

177 "NotExpr", 

178 "OrExpr", 

179 "PaginationSpec", 

180 "PaginatorProtocol", 

181 "ProjectionSpec", 

182 "QueryFilterProtocol", 

183 "SortSpecification", 

184]