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
|
"""Retrieve notifications from Flume API."""
from typing import Any, Dict, Optional
from requests import Session
from .constants import ( # noqa: WPS300
API_BASE_URL,
API_NOTIFICATIONS_URL,
DEFAULT_TIMEOUT,
)
from .utils import configure_logger, flume_response_error # noqa: WPS300
# Configure logging
LOGGER = configure_logger(__name__)
class FlumeNotificationList:
"""Get Flume Notifications list from API."""
def __init__( # noqa: WPS211
self,
flume_auth,
http_session: Optional[Session] = None,
timeout: int = DEFAULT_TIMEOUT,
read: str = "false",
sort_direction: str = "ASC",
) -> None:
"""
Initialize the FlumeNotificationList object.
Args:
flume_auth: Authentication object.
http_session: Optional Requests Session().
timeout: Requests timeout for throttling, default DEFAULT_TIMEOUT.
read: state of notification list, default "false".
sort_direction: Which direction to sort notifications on, default "ASC".
"""
self._timeout = timeout
self._flume_auth = flume_auth
self._read = read
self._sort_direction = sort_direction
self._http_session = http_session or Session()
self.has_next = False
self.next_page = None
self.notification_list = self.get_notifications()
def get_notifications(self) -> Dict[str, Any]:
"""Return all notifications from devices owned by the user.
Returns:
Dict[str, Any]: Notification JSON message from API.
"""
api_url = API_NOTIFICATIONS_URL.format(user_id=self._flume_auth.user_id)
query_string = {
"limit": "50",
"offset": "0",
"sort_direction": self._sort_direction,
"read": self._read,
}
return self._get_notification_request(api_url, query_string)
def get_next_notifications(self):
"""Return next page of notification from devices owned by the user.
Returns:
Returns JSON list of notifications.
Raises:
ValueError: If no next page is available.
"""
if self.has_next:
api_url = f"{API_BASE_URL}{self.next_page}"
query_string = {}
else:
raise ValueError("No next page available.")
return self._get_notification_request(api_url, query_string)
def _has_next_page(self, response_json):
"""Return True if the next page exists.
Args:
response_json (Object): Response from API.
Returns:
Boolean: Returns true if next page exists, False if not.
"""
if response_json is None or response_json.get("pagination") is None:
return False
return (
"next" in response_json["pagination"]
and response_json["pagination"]["next"] is not None
)
def _get_notification_request(self, api_url, query_string):
"""Make an API request to get usage alerts from the Flume API.
Args:
api_url (string): URL for request
query_string (object): query string options
Returns:
object: Reponse in JSON format from API.
"""
response = self._http_session.request(
"GET",
api_url,
headers=self._flume_auth.authorization_header,
params=query_string,
timeout=self._timeout,
)
LOGGER.debug(f"_get_notification_request Response: {response.text}")
# Check for response errors.
flume_response_error("Impossible to retrieve notifications", response)
response_json = response.json()
if self._has_next_page(response_json):
self.next_page = response_json["pagination"]["next"]
self.has_next = True
LOGGER.debug(
f"Next page for Notification results: {self.next_page}",
)
else:
self.has_next = False
self.next_page = None
LOGGER.debug("No further pages for Notification results.")
return response_json["data"]
|
