/
alert_manager.py
226 lines (198 loc) · 9.46 KB
/
alert_manager.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
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
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
from pyowm.constants import ALERT_API_VERSION
from pyowm.commons.http_client import HttpClient
from pyowm.alertapi30.parsers import TriggerParser, AlertParser
from pyowm.alertapi30.uris import TRIGGERS_URI, NAMED_TRIGGER_URI, ALERTS_URI, NAMED_ALERT_URI
from pyowm.utils import timeformatutils
from pyowm.utils import stringutils
class AlertManager:
"""
A manager objects that provides a full interface to OWM Alert API. It implements CRUD methods on Trigger entities
and read/deletion of related Alert objects
:param API_key: the OWM web API key
:type API_key: str
:returns: an *AlertManager* instance
:raises: *AssertionError* when no API Key is provided
"""
def __init__(self, API_key):
assert API_key is not None, 'You must provide a valid API Key'
self.API_key = API_key
self.trigger_parser = TriggerParser()
self.alert_parser = AlertParser()
self.http_client = HttpClient()
def alert_api_version(self):
return ALERT_API_VERSION
# TRIGGER methods
def create_trigger(self, start, end, conditions, area, alert_channels=None):
"""
Create a new trigger on the Alert API with the given parameters
:param start: time object representing the time when the trigger begins to be checked
:type start: int, ``datetime.datetime`` or ISO8601-formatted string
:param end: time object representing the time when the trigger ends to be checked
:type end: int, ``datetime.datetime`` or ISO8601-formatted string
:param conditions: the `Condition` objects representing the set of checks to be done on weather variables
:type conditions: list of `pyowm.utils.alertapi30.Condition` instances
:param area: the geographic are over which conditions are checked: it can be composed by multiple geoJSON types
:type area: list of geoJSON types
:param alert_channels: the alert channels through which alerts originating from this `Trigger` can be consumed.
Defaults to OWM API polling
:type alert_channels: list of `pyowm.utils.alertapi30.AlertChannel` instances
:returns: a *Trigger* instance
:raises: *ValueError* when start or end epochs are `None` or when end precedes start or when conditions or area
are empty collections
"""
assert start is not None
assert end is not None
unix_start = timeformatutils.to_UNIXtime(start)
unix_end = timeformatutils.to_UNIXtime(end)
if unix_start >= unix_end:
raise ValueError("Error: the start epoch must precede the end epoch")
the_time_period = {
"start": {
"expression": "exact",
"amount": unix_start
},
"end": {
"expression": "exact",
"amount": unix_end
}
}
assert conditions is not None
if len(conditions) == 0:
raise ValueError('A trigger must contain at least one condition: you provided none')
the_conditions = [dict(name=c.weather_param, expression=c.operator, amount=c.amount) for c in conditions]
assert area is not None
if len(area) == 0:
raise ValueError('The area for a trigger must contain at least one geoJSON type: you provided none')
the_area = [a.as_dict() for a in area]
# >>> for the moment, no specific handling for alert channels
status, payload = self.http_client.post(
TRIGGERS_URI,
params={'appid': self.API_key},
data=dict(time_period=the_time_period, conditions=the_conditions, area=the_area),
headers={'Content-Type': 'application/json'})
return self.trigger_parser.parse_dict(payload)
def get_triggers(self):
"""
Retrieves all of the user's triggers that are set on the Weather Alert API.
:returns: list of `pyowm.alertapi30.trigger.Trigger` objects
"""
status, data = self.http_client.get_json(
TRIGGERS_URI,
params={'appid': self.API_key},
headers={'Content-Type': 'application/json'})
return [self.trigger_parser.parse_dict(item) for item in data]
def get_trigger(self, trigger_id):
"""
Retrieves the named trigger from the Weather Alert API.
:param trigger_id: the ID of the trigger
:type trigger_id: str
:return: a `pyowm.alertapi30.trigger.Trigger` instance
"""
stringutils.assert_is_string_or_unicode(trigger_id)
status, data = self.http_client.get_json(
NAMED_TRIGGER_URI % trigger_id,
params={'appid': self.API_key},
headers={'Content-Type': 'application/json'})
return self.trigger_parser.parse_dict(data)
def update_trigger(self, trigger):
"""
Updates on the Alert API the trigger record having the ID of the specified Trigger object: the remote record is
updated with data from the local Trigger object.
:param trigger: the Trigger with updated data
:type trigger: `pyowm.alertapi30.trigger.Trigger`
:return: ``None`` if update is successful, an error otherwise
"""
assert trigger is not None
stringutils.assert_is_string_or_unicode(trigger.id)
the_time_period = {
"start": {
"expression": "exact",
"amount": trigger.start
},
"end": {
"expression": "exact",
"amount": trigger.end
}
}
the_conditions = [dict(name=c.weather_param, expression=c.operator, amount=c.amount) for c in trigger.conditions]
the_area = [a.geojson() for a in trigger.area]
status, _ = self.http_client.put(
NAMED_TRIGGER_URI % trigger.id,
params={'appid': self.API_key},
data=dict(time_period=the_time_period, conditions=the_conditions, area=the_area),
headers={'Content-Type': 'application/json'})
def delete_trigger(self, trigger):
"""
Deletes from the Alert API the trigger record identified by the ID of the provided
`pyowm.alertapi30.trigger.Trigger`, along with all related alerts
:param trigger: the `pyowm.alertapi30.trigger.Trigger` object to be deleted
:type trigger: `pyowm.alertapi30.trigger.Trigger`
:returns: `None` if deletion is successful, an exception otherwise
"""
assert trigger is not None
stringutils.assert_is_string_or_unicode(trigger.id)
status, _ = self.http_client.delete(
NAMED_TRIGGER_URI % trigger.id,
params={'appid': self.API_key},
headers={'Content-Type': 'application/json'})
# ALERTS methods
def get_alerts_for(self, trigger):
"""
Retrieves all of the alerts that were fired for the specified Trigger
:param trigger: the trigger
:type trigger: `pyowm.alertapi30.trigger.Trigger`
:return: list of `pyowm.alertapi30.alert.Alert` objects
"""
assert trigger is not None
stringutils.assert_is_string_or_unicode(trigger.id)
status, data = self.http_client.get_json(
ALERTS_URI % trigger.id,
params={'appid': self.API_key},
headers={'Content-Type': 'application/json'})
return [self.alert_parser.parse_dict(item) for item in data]
def get_alert(self, alert_id, trigger):
"""
Retrieves info about the alert record on the Alert API that has the specified ID and belongs to the specified
parent Trigger object
:param trigger: the parent trigger
:type trigger: `pyowm.alertapi30.trigger.Trigger`
:param alert_id: the ID of the alert
:type alert_id `pyowm.alertapi30.alert.Alert`
:return: an `pyowm.alertapi30.alert.Alert` instance
"""
assert trigger is not None
assert alert_id is not None
stringutils.assert_is_string_or_unicode(alert_id)
stringutils.assert_is_string_or_unicode(trigger.id)
status, data = self.http_client.get_json(
NAMED_ALERT_URI % (trigger.id, alert_id),
params={'appid': self.API_key},
headers={'Content-Type': 'application/json'})
return self.alert_parser.parse_dict(data)
def delete_all_alerts_for(self, trigger):
"""
Deletes all of the alert that were fired for the specified Trigger
:param trigger: the trigger whose alerts are to be cleared
:type trigger: `pyowm.alertapi30.trigger.Trigger`
:return: `None` if deletion is successful, an exception otherwise
"""
assert trigger is not None
stringutils.assert_is_string_or_unicode(trigger.id)
status, _ = self.http_client.delete(
ALERTS_URI % trigger.id,
params={'appid': self.API_key},
headers={'Content-Type': 'application/json'})
def delete_alert(self, alert):
"""
Deletes the specified alert from the Alert API
:param alert: the alert to be deleted
:type alert: pyowm.alertapi30.alert.Alert`
:return: ``None`` if the deletion was successful, an error otherwise
"""
assert alert is not None
stringutils.assert_is_string_or_unicode(alert.id)
stringutils.assert_is_string_or_unicode(alert.trigger_id)
status, _ = self.http_client.delete(
NAMED_ALERT_URI % (alert.trigger_id, alert.id),
params={'appid': self.API_key},
headers={'Content-Type': 'application/json'})