1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
|
# ruff: noqa: UP006,UP007
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Generic, List, Optional, TypeVar
from uuid import UUID
__all__ = (
"AbstractAsyncClassicPaginator",
"AbstractAsyncCursorPaginator",
"AbstractAsyncOffsetPaginator",
"AbstractSyncClassicPaginator",
"AbstractSyncCursorPaginator",
"AbstractSyncOffsetPaginator",
"ClassicPagination",
"CursorPagination",
"OffsetPagination",
)
T = TypeVar("T")
C = TypeVar("C", int, str, UUID)
@dataclass
class ClassicPagination(Generic[T]):
"""Container for data returned using limit/offset pagination."""
__slots__ = ("items", "page_size", "current_page", "total_pages")
items: List[T]
"""List of data being sent as part of the response."""
page_size: int
"""Number of items per page."""
current_page: int
"""Current page number."""
total_pages: int
"""Total number of pages."""
@dataclass
class OffsetPagination(Generic[T]):
"""Container for data returned using limit/offset pagination."""
__slots__ = ("items", "limit", "offset", "total")
items: List[T]
"""List of data being sent as part of the response."""
limit: int
"""Maximal number of items to send."""
offset: int
"""Offset from the beginning of the query.
Identical to an index.
"""
total: int
"""Total number of items."""
@dataclass
class CursorPagination(Generic[C, T]):
"""Container for data returned using cursor pagination."""
__slots__ = ("items", "results_per_page", "cursor", "next_cursor")
items: List[T]
"""List of data being sent as part of the response."""
results_per_page: int
"""Maximal number of items to send."""
cursor: Optional[C]
"""Unique ID, designating the last identifier in the given data set.
This value can be used to request the "next" batch of records.
"""
class AbstractSyncClassicPaginator(ABC, Generic[T]):
"""Base paginator class for sync classic pagination.
Implement this class to return paginated result sets using the classic pagination scheme.
"""
@abstractmethod
def get_total(self, page_size: int) -> int:
"""Return the total number of records.
Args:
page_size: Maximal number of records to return.
Returns:
An integer.
"""
raise NotImplementedError
@abstractmethod
def get_items(self, page_size: int, current_page: int) -> list[T]:
"""Return a list of items of the given size 'page_size' correlating with 'current_page'.
Args:
page_size: Maximal number of records to return.
current_page: The current page of results to return.
Returns:
A list of items.
"""
raise NotImplementedError
def __call__(self, page_size: int, current_page: int) -> ClassicPagination[T]:
"""Return a paginated result set.
Args:
page_size: Maximal number of records to return.
current_page: The current page of results to return.
Returns:
A paginated result set.
"""
total_pages = self.get_total(page_size=page_size)
items = self.get_items(page_size=page_size, current_page=current_page)
return ClassicPagination[T](
items=items, total_pages=total_pages, page_size=page_size, current_page=current_page
)
class AbstractAsyncClassicPaginator(ABC, Generic[T]):
"""Base paginator class for async classic pagination.
Implement this class to return paginated result sets using the classic pagination scheme.
"""
@abstractmethod
async def get_total(self, page_size: int) -> int:
"""Return the total number of records.
Args:
page_size: Maximal number of records to return.
Returns:
An integer.
"""
raise NotImplementedError
@abstractmethod
async def get_items(self, page_size: int, current_page: int) -> list[T]:
"""Return a list of items of the given size 'page_size' correlating with 'current_page'.
Args:
page_size: Maximal number of records to return.
current_page: The current page of results to return.
Returns:
A list of items.
"""
raise NotImplementedError
async def __call__(self, page_size: int, current_page: int) -> ClassicPagination[T]:
"""Return a paginated result set.
Args:
page_size: Maximal number of records to return.
current_page: The current page of results to return.
Returns:
A paginated result set.
"""
total_pages = await self.get_total(page_size=page_size)
items = await self.get_items(page_size=page_size, current_page=current_page)
return ClassicPagination[T](
items=items, total_pages=total_pages, page_size=page_size, current_page=current_page
)
class AbstractSyncOffsetPaginator(ABC, Generic[T]):
"""Base paginator class for limit / offset pagination.
Implement this class to return paginated result sets using the limit / offset pagination scheme.
"""
@abstractmethod
def get_total(self) -> int:
"""Return the total number of records.
Returns:
An integer.
"""
raise NotImplementedError
@abstractmethod
def get_items(self, limit: int, offset: int) -> list[T]:
"""Return a list of items of the given size 'limit' starting from position 'offset'.
Args:
limit: Maximal number of records to return.
offset: Starting position within the result set (assume index 0 as starting position).
Returns:
A list of items.
"""
raise NotImplementedError
def __call__(self, limit: int, offset: int) -> OffsetPagination[T]:
"""Return a paginated result set.
Args:
limit: Maximal number of records to return.
offset: Starting position within the result set (assume index 0 as starting position).
Returns:
A paginated result set.
"""
total = self.get_total()
items = self.get_items(limit=limit, offset=offset)
return OffsetPagination[T](items=items, total=total, offset=offset, limit=limit)
class AbstractAsyncOffsetPaginator(ABC, Generic[T]):
"""Base paginator class for limit / offset pagination.
Implement this class to return paginated result sets using the limit / offset pagination scheme.
"""
@abstractmethod
async def get_total(self) -> int:
"""Return the total number of records.
Returns:
An integer.
"""
raise NotImplementedError
@abstractmethod
async def get_items(self, limit: int, offset: int) -> list[T]:
"""Return a list of items of the given size 'limit' starting from position 'offset'.
Args:
limit: Maximal number of records to return.
offset: Starting position within the result set (assume index 0 as starting position).
Returns:
A list of items.
"""
raise NotImplementedError
async def __call__(self, limit: int, offset: int) -> OffsetPagination[T]:
"""Return a paginated result set.
Args:
limit: Maximal number of records to return.
offset: Starting position within the result set (assume index 0 as starting position).
Returns:
A paginated result set.
"""
total = await self.get_total()
items = await self.get_items(limit=limit, offset=offset)
return OffsetPagination[T](items=items, total=total, offset=offset, limit=limit)
class AbstractSyncCursorPaginator(ABC, Generic[C, T]):
"""Base paginator class for sync cursor pagination.
Implement this class to return paginated result sets using the cursor pagination scheme.
"""
@abstractmethod
def get_items(self, cursor: C | None, results_per_page: int) -> tuple[list[T], C | None]:
"""Return a list of items of the size 'results_per_page' following the given cursor, if any,
Args:
cursor: A unique identifier that acts as the 'cursor' after which results should be given.
results_per_page: A maximal number of results to return.
Returns:
A tuple containing the result set and a new cursor that marks the last record retrieved.
The new cursor can be used to ask for the 'next_cursor' batch of results.
"""
raise NotImplementedError
def __call__(self, cursor: C | None, results_per_page: int) -> CursorPagination[C, T]:
"""Return a paginated result set given an optional cursor (unique ID) and a maximal number of results to return.
Args:
cursor: A unique identifier that acts as the 'cursor' after which results should be given.
results_per_page: A maximal number of results to return.
Returns:
A paginated result set.
"""
items, new_cursor = self.get_items(cursor=cursor, results_per_page=results_per_page)
return CursorPagination[C, T](
items=items,
results_per_page=results_per_page,
cursor=new_cursor,
)
class AbstractAsyncCursorPaginator(ABC, Generic[C, T]):
"""Base paginator class for async cursor pagination.
Implement this class to return paginated result sets using the cursor pagination scheme.
"""
@abstractmethod
async def get_items(self, cursor: C | None, results_per_page: int) -> tuple[list[T], C | None]:
"""Return a list of items of the size 'results_per_page' following the given cursor, if any,
Args:
cursor: A unique identifier that acts as the 'cursor' after which results should be given.
results_per_page: A maximal number of results to return.
Returns:
A tuple containing the result set and a new cursor that marks the last record retrieved.
The new cursor can be used to ask for the 'next_cursor' batch of results.
"""
raise NotImplementedError
async def __call__(self, cursor: C | None, results_per_page: int) -> CursorPagination[C, T]:
"""Return a paginated result set given an optional cursor (unique ID) and a maximal number of results to return.
Args:
cursor: A unique identifier that acts as the 'cursor' after which results should be given.
results_per_page: A maximal number of results to return.
Returns:
A paginated result set.
"""
items, new_cursor = await self.get_items(cursor=cursor, results_per_page=results_per_page)
return CursorPagination[C, T](
items=items,
results_per_page=results_per_page,
cursor=new_cursor,
)
|
