forked from tableau/server-client-python
-
Notifications
You must be signed in to change notification settings - Fork 0
/
pager.py
67 lines (53 loc) · 2.66 KB
/
pager.py
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
from functools import partial
from . import RequestOptions
class Pager(object):
"""
Generator that takes an endpoint (top level endpoints with `.get)` and lazily loads items from Server.
Supports all `RequestOptions` including starting on any page. Also used by models to load sub-models
(users in a group, views in a workbook, etc) by passing a different endpoint.
Will loop over anything that returns (List[ModelItem], PaginationItem).
"""
def __init__(self, endpoint, request_opts=None, **kwargs):
if hasattr(endpoint, "get"):
# The simpliest case is to take an Endpoint and call its get
endpoint = partial(endpoint.get, **kwargs)
self._endpoint = endpoint
elif callable(endpoint):
# but if they pass a callable then use that instead (used internally)
endpoint = partial(endpoint, **kwargs)
self._endpoint = endpoint
else:
# Didn't get something we can page over
raise ValueError("Pager needs a server endpoint to page through.")
self._options = request_opts
# If we have options we could be starting on any page, backfill the count
if self._options:
self._count = (self._options.pagenumber - 1) * self._options.pagesize
else:
self._count = 0
self._options = RequestOptions()
def __iter__(self):
# Fetch the first page
current_item_list, last_pagination_item = self._endpoint(self._options)
if last_pagination_item.total_available is None:
# This endpoint does not support pagination, drain the list and return
while current_item_list:
yield current_item_list.pop(0)
return
# Get the rest on demand as a generator
while self._count < last_pagination_item.total_available:
if len(current_item_list) == 0:
current_item_list, last_pagination_item = self._load_next_page(last_pagination_item)
try:
yield current_item_list.pop(0)
self._count += 1
except IndexError:
# The total count on Server changed while fetching exit gracefully
return
def _load_next_page(self, last_pagination_item):
next_page = last_pagination_item.page_number + 1
opts = RequestOptions(pagenumber=next_page, pagesize=last_pagination_item.page_size)
if self._options is not None:
opts.sort, opts.filter = self._options.sort, self._options.filter
current_item_list, last_pagination_item = self._endpoint(opts)
return current_item_list, last_pagination_item