-
Notifications
You must be signed in to change notification settings - Fork 3
/
cursors.py
76 lines (61 loc) · 1.8 KB
/
cursors.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
68
69
70
71
72
73
74
75
76
from __future__ import annotations
from dataclasses import dataclass, make_dataclass
from typing import Generator, List
import requests
# The maximum page size supported by the API.
_MAX_PAGE_SIZE = 500
@dataclass
class CursorPage:
paging: dict
results: List[dict]
class CursorPaginator:
def __init__(
self, session: requests.Session, url: str, params: dict = {}
) -> None:
self.session = session
self.url = url
self.params = params
def fetch_results(self) -> List[dict]:
"""Fetch results.
Collects all results from all pages.
Returns
-------
List[dict]
A coalesced list of all results.
"""
results = []
for page in self.fetch_pages():
results.extend(page.results)
return results
def fetch_pages(self) -> Generator[CursorPage, None, None]:
"""Fetch pages.
Yields
------
Generator[Page, None, None]
"""
next = None
while True:
page = self.fetch_page(next)
yield page
cursors: dict = page.paging.get("cursors", {})
next = cursors.get("next")
if not next:
# stop if a next page is not defined
return
def fetch_page(self, next: str | None = None) -> CursorPage:
"""Fetch a page.
Parameters
----------
next : str | None, optional
the next page identifier or None to fetch the first page, by default None
Returns
-------
Page
"""
params = {
**self.params,
"next": next,
"limit": _MAX_PAGE_SIZE,
}
response = self.session.get(self.url, params=params)
return CursorPage(**response.json())