-
Notifications
You must be signed in to change notification settings - Fork 55
Expand file tree
/
Copy pathcommon_objects.py
More file actions
54 lines (42 loc) · 1.76 KB
/
common_objects.py
File metadata and controls
54 lines (42 loc) · 1.76 KB
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
import json
from collections.abc import Callable
from typing import TypeVar, Generic
class Request(dict):
def __setitem__(self, key, value):
if value is not None:
super(Request, self).__setitem__(key, value)
class JsonObject(dict):
def __init__(self, *args, **kwargs):
super(JsonObject, self).__init__(*args, **kwargs)
json = json.dumps
ENTITY = TypeVar('ENTITY')
class Pagination(Generic[ENTITY]):
def __init__(self, first_page: JsonObject,
total_result: int,
next_page_loader: Callable[[JsonObject], JsonObject | None],
resources_accessor: Callable[[JsonObject], list[JsonObject]],
instance_creator: Callable[[JsonObject], ENTITY]):
self._first_page = first_page
self._total_results = total_result
self._next_page_loader = next_page_loader
self._resources_accessor = resources_accessor
self._instance_creator = instance_creator
self._cursor = None
self._current_page = None
@property
def total_results(self) -> int:
return self._total_results
def __iter__(self):
return self
def __next__(self) -> ENTITY:
try:
if self._cursor is None:
self._current_page = self._first_page
self._cursor = self._resources_accessor(self._current_page).__iter__()
return self._instance_creator(self._cursor.__next__())
except StopIteration:
self._current_page = self._next_page_loader(self._current_page)
if self._current_page is None:
raise
self._cursor = self._resources_accessor(self._current_page).__iter__()
return self._instance_creator(self._cursor.__next__())