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
|
import json
from typing import Any, Dict, Optional
from twilio.base.exceptions import TwilioException
from twilio.http.response import Response
class Page(object):
"""
Represents a page of records in a collection.
A `Page` lets you iterate over its records and fetch the next and previous
pages in the collection.
"""
META_KEYS = {
"end",
"first_page_uri",
"next_page_uri",
"last_page_uri",
"page",
"page_size",
"previous_page_uri",
"total",
"num_pages",
"start",
"uri",
}
def __init__(self, version, response: Response, solution={}):
payload = self.process_response(response)
self._version = version
self._payload = payload
self._solution = solution
self._records = iter(self.load_page(payload))
def __iter__(self):
"""
A `Page` is a valid iterator.
"""
return self
def __next__(self):
return self.next()
def next(self):
"""
Returns the next record in the `Page`.
"""
return self.get_instance(next(self._records))
@classmethod
def process_response(cls, response: Response) -> Any:
"""
Load a JSON response.
:param response: The HTTP response.
:return The JSON-loaded content.
"""
if response.status_code != 200:
raise TwilioException("Unable to fetch page", response)
return json.loads(response.text)
def load_page(self, payload: Dict[str, Any]):
"""
Parses the collection of records out of a list payload.
:param payload: The JSON-loaded content.
:return list: The list of records.
"""
if "meta" in payload and "key" in payload["meta"]:
return payload[payload["meta"]["key"]]
else:
keys = set(payload.keys())
key = keys - self.META_KEYS
if len(key) == 1:
return payload[key.pop()]
if "Resources" in payload:
return payload["Resources"]
raise TwilioException("Page Records can not be deserialized")
@property
def previous_page_url(self) -> Optional[str]:
"""
:return str: Returns a link to the previous_page_url or None if doesn't exist.
"""
if "meta" in self._payload and "previous_page_url" in self._payload["meta"]:
return self._payload["meta"]["previous_page_url"]
elif (
"previous_page_uri" in self._payload and self._payload["previous_page_uri"]
):
return self._version.domain.absolute_url(self._payload["previous_page_uri"])
return None
@property
def next_page_url(self) -> Optional[str]:
"""
:return str: Returns a link to the next_page_url or None if doesn't exist.
"""
if "meta" in self._payload and "next_page_url" in self._payload["meta"]:
return self._payload["meta"]["next_page_url"]
elif "next_page_uri" in self._payload and self._payload["next_page_uri"]:
return self._version.domain.absolute_url(self._payload["next_page_uri"])
return None
def get_instance(self, payload: Dict[str, Any]) -> Any:
"""
:param dict payload: A JSON-loaded representation of an instance record.
:return: A rich, resource-dependent object.
"""
raise TwilioException(
"Page.get_instance() must be implemented in the derived class"
)
def next_page(self) -> Optional["Page"]:
"""
Return the `Page` after this one.
:return The next page.
"""
if not self.next_page_url:
return None
response = self._version.domain.twilio.request("GET", self.next_page_url)
cls = type(self)
return cls(self._version, response, self._solution)
async def next_page_async(self) -> Optional["Page"]:
"""
Asynchronously return the `Page` after this one.
:return The next page.
"""
if not self.next_page_url:
return None
response = await self._version.domain.twilio.request_async(
"GET", self.next_page_url
)
cls = type(self)
return cls(self._version, response, self._solution)
def previous_page(self) -> Optional["Page"]:
"""
Return the `Page` before this one.
:return The previous page.
"""
if not self.previous_page_url:
return None
response = self._version.domain.twilio.request("GET", self.previous_page_url)
cls = type(self)
return cls(self._version, response, self._solution)
async def previous_page_async(self) -> Optional["Page"]:
"""
Asynchronously return the `Page` before this one.
:return The previous page.
"""
if not self.previous_page_url:
return None
response = await self._version.domain.twilio.request_async(
"GET", self.previous_page_url
)
cls = type(self)
return cls(self._version, response, self._solution)
def __repr__(self) -> str:
return "<Page>"
|
