Coverage for src/lexigram/graphql/pagination/helpers.py: 14%

1"""Utility routines for GraphQL pagination. 

2 

3This module hosts the cursor encoding/decoding logic and pagination helper 

4functions. Previously these were in the package root alongside the data 

5classes; they have now been moved here to keep ``__init__`` lean. 

6""" 

7 

8from __future__ import annotations 

9 

10import base64 

11import binascii 

12from typing import TYPE_CHECKING, TypeVar 

13 

14from lexigram.graphql.pagination.types import ( 

15 CursorConnection, 

16 CursorPaginationInput, 

17 Edge, 

18 OffsetPaginationInput, 

19 PageInfo, 

20 PaginationResult, 

21) 

22 

23if TYPE_CHECKING: 

24 from collections.abc import Sequence 

25 

26# Cursor helpers ------------------------------------------------------------- 

27 

28T = TypeVar("T") 

29 

30 

31def encode_cursor(offset: int) -> str: 

32 """Encode an offset as a base64 cursor. 

33 

34 Args: 

35 offset: The offset to encode. 

36 

37 Returns: 

38 Base64-encoded cursor string. 

39 """ 

40 cursor = f"cursor:{offset}" 

41 return base64.b64encode(cursor.encode()).decode() 

42 

43 

44def decode_cursor(cursor: str) -> int: 

45 """Decode a base64 cursor to an offset. 

46 

47 Args: 

48 cursor: The cursor string to decode. 

49 

50 Returns: 

51 The decoded offset. 

52 

53 Raises: 

54 ValueError: If the cursor is invalid. 

55 """ 

56 try: 

57 decoded = base64.b64decode(cursor.encode()).decode() 

58 if decoded.startswith("cursor:"): 

59 return int(decoded[7:]) 

60 except (ValueError, binascii.Error): 

61 pass 

62 raise ValueError(f"Invalid cursor: {cursor}") 

63 

64 

65def encode_cursor_from_id(item_id: str) -> str: 

66 """Encode an item ID as an opaque cursor. 

67 

68 Args: 

69 item_id: The item ID to encode. 

70 

71 Returns: 

72 Base64-encoded cursor string. 

73 """ 

74 cursor = f"id:{item_id}" 

75 return base64.b64encode(cursor.encode()).decode() 

76 

77 

78def decode_cursor_to_id(cursor: str) -> str: 

79 """Decode a cursor to an item ID. 

80 

81 Args: 

82 cursor: The cursor string to decode. 

83 

84 Returns: 

85 The decoded item ID. 

86 

87 Raises: 

88 ValueError: If the cursor is invalid. 

89 """ 

90 try: 

91 decoded = base64.b64decode(cursor.encode()).decode() 

92 if decoded.startswith("id:"): 

93 return decoded[3:] 

94 except (ValueError, binascii.Error): 

95 pass 

96 raise ValueError(f"Invalid cursor: {cursor}") 

97 

98 

99# Pagination implementations ------------------------------------------------ 

100 

101 

102async def paginate_connection( 

103 items: Sequence[T], 

104 pagination: CursorPaginationInput | dict, 

105 total_count: int | None = None, 

106) -> CursorConnection[T]: 

107 """Create a Relay Connection from a list of items. 

108 

109 Args: 

110 items: Sequence of items to paginate. 

111 pagination: Cursor pagination input. 

112 total_count: Total count of items (if known). 

113 

114 Returns: 

115 A Connection with edges and page info. 

116 """ 

117 if isinstance(pagination, dict): 

118 pagination = CursorPaginationInput(**pagination) 

119 

120 pagination.validate() 

121 

122 # Calculate offsets 

123 start_offset = 0 

124 end_offset = len(items) 

125 

126 if pagination.after: 

127 try: 

128 start_offset = decode_cursor(pagination.after) + 1 

129 except ValueError: 

130 start_offset = 0 

131 

132 if pagination.before: 

133 try: 

134 end_offset = decode_cursor(pagination.before) 

135 except ValueError: 

136 end_offset = len(items) 

137 

138 # Apply first/last limits 

139 if pagination.first is not None: 

140 end_offset = min(start_offset + pagination.first, end_offset) 

141 elif pagination.last is not None: 

142 start_offset = max(end_offset - pagination.last, start_offset) 

143 

144 # Get the slice 

145 page_items = items[start_offset:end_offset] 

146 

147 # Calculate total if not provided 

148 if total_count is None: 

149 total_count = len(items) 

150 

151 # Create edges 

152 edges = [] 

153 for i, item in enumerate(page_items): 

154 offset = start_offset + i 

155 edge = Edge( 

156 node=item, 

157 cursor=encode_cursor(offset), 

158 ) 

159 edges.append(edge) 

160 

161 # Calculate has_next/has_previous 

162 has_next = end_offset < len(items) 

163 has_previous = start_offset > 0 

164 

165 page_info = PageInfo( 

166 has_next_page=has_next, 

167 has_previous_page=has_previous, 

168 start_cursor=edges[0].cursor if edges else None, 

169 end_cursor=edges[-1].cursor if edges else None, 

170 ) 

171 

172 return CursorConnection( 

173 edges=edges, 

174 page_info=page_info, 

175 ) 

176 

177 

178async def paginate_offset( 

179 items: Sequence[T], 

180 pagination: OffsetPaginationInput | dict, 

181 total_count: int | None = None, 

182) -> PaginationResult[T]: 

183 """Paginate using offset-based pagination. 

184 

185 Args: 

186 items: Sequence of items to paginate. 

187 pagination: Offset pagination input. 

188 total_count: Total count of items (if known). 

189 

190 Returns: 

191 PaginationResult with items and metadata. 

192 """ 

193 if isinstance(pagination, dict): 

194 pagination = OffsetPaginationInput(**pagination) 

195 

196 pagination.validate() 

197 

198 # Calculate total if not provided 

199 if total_count is None: 

200 total_count = len(items) 

201 

202 # Apply offset and limit 

203 offset = pagination.offset 

204 limit = min(pagination.offset + pagination.limit, total_count) 

205 

206 page_items = items[offset:limit] 

207 

208 # Calculate has_next/has_previous 

209 has_next = limit < total_count 

210 has_previous = offset > 0 

211 

212 return PaginationResult( 

213 items=list(page_items), 

214 total_count=total_count, 

215 has_next=has_next, 

216 has_previous=has_previous, 

217 ) 

218 

219 

220def calculate_offset( 

221 first: int | None, 

222 after: str | None, 

223 last: int | None, 

224 before: str | None, 

225 default_first: int = 10, 

226) -> tuple[int, int]: 

227 """Calculate offset and limit from cursor pagination inputs. 

228 

229 Args: 

230 first: Number of items forward. 

231 after: Cursor to start after. 

232 last: Number of items backward. 

233 before: Cursor to start before. 

234 default_first: Default number of items if not specified. 

235 

236 Returns: 

237 Tuple of (offset, limit). 

238 """ 

239 offset = 0 

240 limit = default_first 

241 

242 # Determine direction 

243 if after: 

244 try: 

245 offset = decode_cursor(after) + 1 

246 except ValueError: 

247 offset = 0 

248 

249 if before: 

250 try: 

251 offset = decode_cursor(before) - default_first 

252 offset = max(offset, 0) 

253 except ValueError: 

254 pass 

255 

256 if first is not None: 

257 limit = first 

258 elif last is not None: 

259 limit = last 

260 

261 return offset, limit