This repository has been archived by the owner on Jul 17, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 8
/
subscriber.py
523 lines (423 loc) · 16.5 KB
/
subscriber.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
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
"""Manage Subcribers."""
from warnings import warn
import mailerlite.client as client
from mailerlite.constants import Subscriber, Activity, Group, Field
def get_id_or_email_identifier(**kwargs):
"""Return id or email identifier."""
identifier = None
for k in ['id', 'email']:
identifier = kwargs.get(k, None)
if identifier is not None:
break
return identifier
class Subscribers:
def __init__(self, headers):
"""Initialize Subscribers object.
Parameters
----------
headers : dict
request header containing your mailerlite api_key.
More information : https://developers.mailerlite.com/docs/request
"""
valid_headers, error_msg = client.check_headers(headers)
if not valid_headers:
raise ValueError(error_msg)
self.headers = headers
def active(self, limit=100, offset=0, as_json=False):
"""Get all active Subscribers from your account.
look at https://developers.mailerlite.com/v2/reference#subscribers
Parameters
----------
limit : int
How many subscribers you want
offset : int
page index
as_json : bool
return result as json format
Returns
-------
subscribers: list
all active Subscribers. More informations :
https://developers.mailerlite.com/v2/reference#subscribers
"""
return self.all(limit=limit, offset=offset, stype='active',
as_json=as_json)
def unsubscribed(self, limit=100, offset=0, as_json=False):
"""Get all unsubscribed Subscribers from your account.
look at https://developers.mailerlite.com/v2/reference#subscribers
Parameters
----------
limit : int
How many subscribers you want
offset : int
page index
as_json : bool
return result as json format
Returns
-------
subscribers: list
all unsubscribed Subscribers. More informations :
https://developers.mailerlite.com/v2/reference#subscribers
"""
return self.all(limit=limit, offset=offset, stype='unsubscribed',
as_json=as_json)
def bounced(self, limit=100, offset=0, as_json=False):
"""Get all bounced Subscribers from your account.
look at https://developers.mailerlite.com/v2/reference#subscribers
Parameters
----------
limit : int
How many subscribers you want
offset : int
page index
as_json : bool
return result as json format
Returns
-------
subscribers: list
all bounced Subscribers. More informations :
https://developers.mailerlite.com/v2/reference#subscribers
"""
return self.all(limit=limit, offset=offset, stype='bounced',
as_json=as_json)
def junk(self, limit=100, offset=0, as_json=False):
"""Get all junk Subscribers from your account.
look at https://developers.mailerlite.com/v2/reference#subscribers
Parameters
----------
limit : int
How many subscribers you want
offset : int
page index
as_json : bool
return result as json format
Returns
-------
subscribers: list
all junk Subscribers. More informations :
https://developers.mailerlite.com/v2/reference#subscribers
"""
return self.all(limit=limit, offset=offset, stype='junk',
as_json=as_json)
def unconfirmed(self, limit=100, offset=0, as_json=False):
"""Get all unconfirmed Subscribers from your account.
look at https://developers.mailerlite.com/v2/reference#subscribers
Parameters
----------
limit : int
How many subscribers you want
offset : int
page index
as_json : bool
return result as json format
Returns
-------
subscribers: list
all unconfirmed Subscribers. More informations :
https://developers.mailerlite.com/v2/reference#subscribers
"""
return self.all(limit=limit, offset=offset, stype='unconfirmed',
as_json=as_json)
def all(self, limit=100, offset=0, stype=None, as_json=False):
"""Get paginated details of all Subscribers from your account.
look at https://developers.mailerlite.com/v2/reference#subscribers
Parameters
----------
limit : int
How many subscribers you want
offset : int
page index
stype : str
Define subscriber type: Here are the possible values:
* None - All subscribers (default)
* active
* unsubscribed
* bounced
* junk
* unconfirmed
as_json : bool
return result as json format
Returns
-------
subscribers: list
all desired Subscribers. More informations :
https://developers.mailerlite.com/v2/reference#subscribers
"""
params = {'limit': limit, 'offset': offset}
if stype and stype.lower() in ['active', 'unsubscribed', 'bounced',
'junk', 'unconfirmed']:
params.update({'type': stype})
url = client.build_url('subscribers', **params)
_, res_json = client.get(url, headers=self.headers)
if as_json or not res_json:
return res_json
for res in res_json:
res['fields'] = [Field(**field) for field in res['fields']]
all_subscribers = [Subscriber(**res) for res in res_json]
return all_subscribers
def count(self, stype=None, as_json=False):
"""Get the count of subscribers of a type.
Please, be aware that `count` is not documented in the official API
https://developers.mailerlite.com/reference#subscribers
Parameters
----------
stype : str
Define subscriber type: Here are the possible values:
* None - All subscribers (default)
* active
* unsubscribed
* bounced
* junk
* unconfirmed
as_json : bool
return result as json format
Returns
-------
number: Integer
the count of subscribers
"""
warn("Please, be aware that `count` is not in the official API")
params = {}
if stype and stype.lower() in ['active', 'unsubscribed', 'bounced',
'junk', 'unconfirmed']:
params.update({'type': stype})
url = client.build_url('subscribers', 'count', **params)
_, res_json = client.get(url, headers=self.headers)
if as_json or not res_json:
return res_json
return res_json['count']
def get(self, as_json=False, **identifier):
"""Get a single subscriber from your account.
https://developers.mailerlite.com/v2/reference#single-subscriber
Parameters
----------
identifier : str
should be subscriber id or email.
e.g: id=1343965485 or email='[email protected]'
as_json : bool
return result as json format
Returns
-------
subscriber: :class:Subscriber
a single subscriber
"""
path = get_id_or_email_identifier(**identifier)
if path is None:
raise IOError('An identifier must be define')
url = client.build_url('subscribers', path)
_, res_json = client.get(url, headers=self.headers)
if as_json or not res_json:
return res_json
res_json['fields'] = [Field(**res) for res in res_json['fields']]
return Subscriber(**res_json)
def delete(self, subscriber_id):
"""Remove a subscribers.
Parameters
----------
subscriber_id : int
subscribers id
Returns
-------
success: bool
deletion status
"""
url = client.build_url('subscribers', subscriber_id)
return client.delete(url, headers=self.headers)
def search(self, search=None, limit=100, offset=0, minimized=True,
as_json=False):
"""Get paginated details of all Subscribers from your account.
look at https://developers.mailerlite.com/v2/reference#subscribers
Parameters
----------
search : str
query parameter to search
limit : int
How many subscribers you want
offset : int
page index
minimized : bool
return minimized response with: id, email, type
default: True
as_json : bool
return result as json format
Returns
-------
subscribers: list
all desired Subscribers. More informations :
https://developers.mailerlite.com/v2/reference#subscribers
"""
params = {'limit': limit, 'offset': offset, 'minimized': minimized}
if search is not None:
params.update({'query': search})
url = client.build_url('subscribers', 'search', **params)
_, res_json = client.get(url, headers=self.headers)
if as_json or not res_json:
return res_json
if not minimized:
for res in res_json:
res['fields'] = [Field(**field) for field in res['fields']]
all_subscribers = [Subscriber(**res) for res in res_json]
return all_subscribers
def groups(self, as_json=False, **identifier):
"""Get groups subscriber belongs to.
More informations:
https://developers.mailerlite.com/v2/reference#groups-subscriber-belongs-to
Parameters
----------
identifier : str
should be subscriber id or email.
e.g: id=1343965485 or email='[email protected]'
as_json : bool
return result as json format
Returns
-------
groups: list
all groups that a subscriber belongs to. More informations :
https://developers.mailerlite.com/v2/reference#groups
"""
path = get_id_or_email_identifier(**identifier)
if path is None:
raise IOError('An identifier must be define')
url = client.build_url('subscribers', path, 'groups')
_, res_json = client.get(url, headers=self.headers)
if as_json or not res_json:
return res_json
all_groups = [Group(**res) for res in res_json]
return all_groups
def activity(self, as_json=False, atype=None, limit=100,
offset=0, **identifier):
"""Get activities (clicks, opens, etc) of selected subscriber.
More informations:
https://developers.mailerlite.com/v2/reference#activity-of-single-subscriber
Parameters
----------
identifier : str
should be subscriber id or email.
e.g: id=1343965485 or email='[email protected]'
as_json : bool
return result as json format
atype : str
Define activity type: Here are the possible values:
* None - All activities (default)
* opens
* clicks
* bounces
* junks
* unsubscribes
* forwards
* sendings
limit : int, optional
How many subscribers you want, default 100
offset : int, optional
page index, default 0
Returns
-------
activities: list
all subscriber activities. More informations :
https://developers.mailerlite.com/v2/reference#activity-of-single-subscriber
"""
params = {'limit': limit, 'offset': offset}
path = get_id_or_email_identifier(**identifier)
if path is None:
raise IOError('An identifier must be define')
args = ['subscribers', path, 'activity']
if atype:
possible_atype = ['opens', 'clicks', 'junks', 'bounces',
'unsubscribes', 'forwards', 'sendings']
if atype not in possible_atype:
raise ValueError('Incorrect value atype. Activity type should'
' be {0}'.format(possible_atype))
args.append(atype)
url = client.build_url(*args, **params)
_, res_json = client.get(url, headers=self.headers)
if as_json or not res_json:
return res_json
all_activities = [Activity(**res) for res in res_json]
return all_activities
def update(self, data, as_json=False, **identifier):
"""Update single subscriber.
https://developers.mailerlite.com/v2/reference#update-subscriber
Parameters
----------
data : dict
subscriber object. only the email is required.
you can use the following example:
data = {'name' : 'John',
'fields' : {'company': 'MailerLite'}
}
as_json : bool
return result as json format
identifier : str
should be subscriber id or email.
e.g: id=1343965485 or email='[email protected]'
Returns
-------
response : int
response value
content : dict
The JSON output from the API
Notes
-----
The email of a subscriber can not be updated
"""
path = get_id_or_email_identifier(**identifier)
if path is None:
raise IOError('An identifier must be define')
if 'email' in data.keys():
raise ValueError("Subscriber email can not be updated. Please, "
"remove this field or create a new Subscriber. "
"For more informations, look at "
"http://help.mailerlite.com/article/show"
"/29233-how-to-edit-a-subscribers-data")
optional_keys = ['name', 'type', 'fields', 'resend_autoresponders']
unknown_keys = [d for d in data.keys() if d not in optional_keys
if d not in ['groups', 'segments']]
if unknown_keys:
raise ValueError("The following keys are unknown: {}"
.format(unknown_keys))
url = client.build_url('subscribers', path)
_, res_json = client.put(url, body=data, headers=self.headers)
if as_json or not res_json:
return res_json
res_json['fields'] = [Field(**res) for res in res_json['fields']]
return Subscriber(**res_json)
def create(self, data, as_json=False):
"""Add new single subscriber.
https://developers.mailerlite.com/v2/reference#create-a-subscriber
Parameters
----------
data : dict
subscriber object. only the email is required.
you can use the following example:
data = {'name' : 'John',
'email' : '[email protected]',
'fields' : {'company': 'MailerLite'}
}
as_json : bool
return result as json format
Returns
-------
subscriber: :class:Subscriber
a single subscriber
"""
if not isinstance(data, dict):
raise ValueError('In data should be a dictionary.')
required_keys = ['email', ]
optional_keys = ['name', 'fields', 'resubscribe', 'type',
'signup_ip', 'signup_timestamp', 'confirmation_ip',
'confirmation_timestamp']
available_keys = required_keys + optional_keys
errors = [rk for rk in required_keys if rk not in data.keys()]
if errors:
raise ValueError("The following keys are missing and they"
" are required : {}".format(errors))
unknown_keys = [d for d in data.keys() if d not in available_keys
if d not in ['groups', 'segments']]
if unknown_keys:
raise ValueError("The following keys are unknown: {}"
.format(unknown_keys))
url = client.build_url('subscribers')
_, res_json = client.post(url, body=data, headers=self.headers)
if as_json or not res_json:
return res_json
return Subscriber(**res_json)