FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
python-client/splitio/push/parser.py at development · splitio/python-client · GitHub
Uh oh!
There was an error while loading.
Please reload this page
.
splitio
/
python-client
Public
Notifications
You must be signed in to change notification settings
Fork
13
Star
19
Code
Issues
4
Pull requests
6
Actions
Projects
Security and quality
0
Insights
Additional navigation options
Code
Issues
Pull requests
Actions
Projects
Security and quality
Insights
Expand file tree
Breadcrumbs
python-client
/
splitio
/
push
/
parser.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
638 lines (488 loc) · 16.7 KB
Breadcrumbs
python-client
/
splitio
/
push
/
parser.py
Copy path
File metadata and controls
638 lines (488 loc) · 16.7 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
"""SSE Notification definitions."""
import
abc
import
json
from
enum
import
Enum
from
splitio
.
util
.
decorators
import
abstract_property
from
splitio
.
util
.
time
import
utctime_ms
from
splitio
.
push
.
sse
import
SSE_EVENT_ERROR
,
SSE_EVENT_MESSAGE
class
EventType
(
Enum
):
"""Event type enumeration."""
MESSAGE
=
SSE_EVENT_MESSAGE
ERROR
=
SSE_EVENT_ERROR
class
MessageType
(
Enum
):
"""Message type enumeration."""
UPDATE
=
0
OCCUPANCY
=
1
CONTROL
=
2
class
UpdateType
(
Enum
):
"""Message type enumeration."""
SPLIT_UPDATE
=
'SPLIT_UPDATE'
SPLIT_KILL
=
'SPLIT_KILL'
SEGMENT_UPDATE
=
'SEGMENT_UPDATE'
RB_SEGMENT_UPDATE
=
'RB_SEGMENT_UPDATE'
class
ControlType
(
Enum
):
"""Control type enumeration."""
STREAMING_ENABLED
=
'STREAMING_ENABLED'
STREAMING_PAUSED
=
'STREAMING_PAUSED'
STREAMING_DISABLED
=
'STREAMING_DISABLED'
TAG_OCCUPANCY
=
'[meta]occupancy'
class
EventParsingException
(
Exception
):
"""Exception to be raised on parser errors."""
pass
class
BaseEvent
(
object
,
metaclass
=
abc
.
ABCMeta
):
"""Base event that reqiures subclasses tu have a type."""
@
abstract_property
def
event_type
(
self
):
# pylint:disable=no-self-use
"""
Return the event type.
:returns: The type of this parsed event.
:rtype: EventType
"""
pass
class
AblyError
(
BaseEvent
):
"""Ably Error message."""
def
__init__
(
self
,
code
,
status_code
,
message
,
href
):
"""
Class constructor.
:param code: error code
:type code: int
:param status_code: http status cude
:type status_code: int
:param message: error message
:type message: str
:param href: link to error description
:type href: str
"""
self
.
_code
=
code
self
.
_status_code
=
status_code
self
.
_message
=
message
self
.
_href
=
href
self
.
_timestamp
=
utctime_ms
()
@
property
def
event_type
(
self
):
# pylint:disable=no-self-use
"""
Return the event type.
:returns: The type of this parsed event.
:rtype: MessageType
"""
return
EventType
.
ERROR
@
property
def
code
(
self
):
"""
Return the error code.
:returns: ably error code.
:rtype: int
"""
return
self
.
_code
@
property
def
status_code
(
self
):
"""
Return the http status code.
:returns: http status error code.
:rtype: int
"""
return
self
.
_status_code
@
property
def
message
(
self
):
"""
Return the ably error message.
:returns: ably error message.
:rtype: str
"""
return
self
.
_message
@
property
def
href
(
self
):
"""
Return the link of the error description.
:returns: error description url
:rtype: str
"""
return
self
.
_href
@
property
def
timestamp
(
self
):
"""
Return a the timestamp when this error was constructed.
:returns: approximate error timestamp
:rtype: int
"""
return
self
.
_timestamp
def
should_be_ignored
(
self
):
"""
Return whether this error should be ignored or not.
:returns: True if this error should be ignored. False otherwise.
:rtype: bool
"""
return
self
.
_code
<
40000
or
self
.
_code
>
49999
def
is_retryable
(
self
):
"""
Return whether this error is retryable or not.
:returns: True if this error is retryable. False otherwise.
:rtype: bool
"""
return
self
.
_code
>=
40140
and
self
.
_code
<=
40149
def
__str__
(
self
):
"""Return string representation."""
return
"AblyError - code=%d, status=%d, message=%s, href=%s"
%
\
(
self
.
code
,
self
.
status_code
,
self
.
message
,
self
.
href
)
class
BaseMessage
(
BaseEvent
,
metaclass
=
abc
.
ABCMeta
):
"""Message type event."""
def
__init__
(
self
,
channel
,
timestamp
):
"""
Construct a message's base structure.
:param channel: channel where the notification was received.
:type channel: str
"""
self
.
_channel
=
channel
self
.
_timestamp
=
timestamp
@
property
def
channel
(
self
):
"""
Return the channel where the message arrived.
:returns: channel
:rtype: str
"""
return
self
.
_channel
@
property
def
timestamp
(
self
):
"""
Return the timestamp when the message was sent.
:returns: message sending timestamp
:rtype: int
"""
return
self
.
_timestamp
@
property
def
event_type
(
self
):
# pylint:disable=no-self-use
"""
Return the event type.
:returns: The type of this parsed event.
:rtype: MessageType
"""
return
EventType
.
MESSAGE
@
abstract_property
def
message_type
(
self
):
# pylint:disable=no-self-use
"""
Return the message type.
:returns: The type of this parsed Message.
:rtype: MessageType
"""
pass
class
OccupancyMessage
(
BaseMessage
):
"""Ably publisher occupancy notification."""
def
__init__
(
self
,
channel
,
timestamp
,
publishers
):
"""
Construct an occupancy message.
:param channel: channel where occupancy is being announced.
:type channel: str
:param publishers: number of active publishers attached to this channel.
:type data: int
"""
BaseMessage
.
__init__
(
self
,
channel
,
timestamp
)
self
.
_publishers
=
publishers
@
property
def
message_type
(
self
):
# pylint:disable=no-self-use
"""
Return the message type.
:returns: The type of this parsed Message.
:rtype: MessageType
"""
return
MessageType
.
OCCUPANCY
@
property
def
channel
(
self
):
"""
Return the channel on which this message was received.
:returns: channel name
:rtype: str
"""
return
self
.
_channel
.
replace
(
'[?occupancy=metrics.publishers]'
,
''
)
@
property
def
publishers
(
self
):
"""
Return the number of publishers of this channel.
:returns: attahed publisher count.
:rtype: int
"""
return
self
.
_publishers
def
__str__
(
self
):
"""Return string representation."""
return
"Occupancy - channel=%s, publishers=%d"
%
(
self
.
channel
,
self
.
publishers
)
class
BaseUpdate
(
BaseMessage
,
metaclass
=
abc
.
ABCMeta
):
"""Feature flag data update notification."""
def
__init__
(
self
,
channel
,
timestamp
,
change_number
):
"""
Construct an update event.
:param data: raw message data.
:type data: dict
:param channel: channel where the message came from.
:type channel: str
"""
BaseMessage
.
__init__
(
self
,
channel
,
timestamp
)
self
.
_change_number
=
change_number
@
abstract_property
def
update_type
(
self
):
# pylint:disable=no-self-use
"""
Return the message type.
:returns: The type of this parsed Update.
:rtype: UpdateType
"""
pass
@
property
def
message_type
(
self
):
# pylint:disable=no-self-use
"""
Return the message type.
:returns: The type of this parsed event.
:rtype: MessageType
"""
return
MessageType
.
UPDATE
@
property
def
change_number
(
self
):
"""
Return the change number associated with the data update.
:returns: change number
:rtype: int
"""
return
self
.
_change_number
class
SplitChangeUpdate
(
BaseUpdate
):
"""Feature flag Change notification."""
def
__init__
(
self
,
channel
,
timestamp
,
change_number
,
previous_change_number
,
feature_flag_definition
,
compression
):
"""Class constructor."""
BaseUpdate
.
__init__
(
self
,
channel
,
timestamp
,
change_number
)
self
.
_previous_change_number
=
previous_change_number
self
.
_object_definition
=
feature_flag_definition
self
.
_compression
=
compression
@
property
def
update_type
(
self
):
# pylint:disable=no-self-use
"""
Return the message type.
:returns: The type of this parsed Update.
:rtype: UpdateType
"""
return
UpdateType
.
SPLIT_UPDATE
@
property
def
previous_change_number
(
self
):
# pylint:disable=no-self-use
"""
Return previous change number
:returns: The previous change number
:rtype: int
"""
return
self
.
_previous_change_number
@
property
def
object_definition
(
self
):
# pylint:disable=no-self-use
"""
Return feature flag definition
:returns: The new feature flag definition
:rtype: str
"""
return
self
.
_object_definition
@
property
def
compression
(
self
):
# pylint:disable=no-self-use
"""
Return previous compression type
:returns: The compression type
:rtype: int
"""
return
self
.
_compression
def
__str__
(
self
):
"""Return string representation."""
return
"SplitChange - changeNumber=%d"
%
(
self
.
change_number
)
class
SplitKillUpdate
(
BaseUpdate
):
"""Feature flag Kill notification."""
def
__init__
(
self
,
channel
,
timestamp
,
change_number
,
feature_flag_name
,
default_treatment
):
# pylint:disable=too-many-arguments
"""Class constructor."""
BaseUpdate
.
__init__
(
self
,
channel
,
timestamp
,
change_number
)
self
.
_feature_flag_name
=
feature_flag_name
self
.
_default_treatment
=
default_treatment
@
property
def
update_type
(
self
):
# pylint:disable=no-self-use
"""
Return the message type.
:returns: The type of this parsed Update.
:rtype: UpdateType
"""
return
UpdateType
.
SPLIT_KILL
@
property
def
feature_flag_name
(
self
):
"""
Return the name of the killed feature flag.
:returns: name of the killed feature flag
:rtype: str
"""
return
self
.
_feature_flag_name
@
property
def
default_treatment
(
self
):
"""
Return the default treatment.
:returns: default treatment
:rtype: str
"""
return
self
.
_default_treatment
def
__str__
(
self
):
"""Return string representation."""
return
"SplitKill - changeNumber=%d, name=%s, defaultTreatment=%s"
%
\
(
self
.
change_number
,
self
.
feature_flag_name
,
self
.
default_treatment
)
class
SegmentChangeUpdate
(
BaseUpdate
):
"""Segment Change notification."""
def
__init__
(
self
,
channel
,
timestamp
,
change_number
,
segment_name
):
"""Class constructor."""
BaseUpdate
.
__init__
(
self
,
channel
,
timestamp
,
change_number
)
self
.
_segment_name
=
segment_name
@
property
def
update_type
(
self
):
# pylint:disable=no-self-use
"""
Return the message type.
:returns: The type of this parsed Update.
:rtype: UpdateType
"""
return
UpdateType
.
SEGMENT_UPDATE
@
property
def
segment_name
(
self
):
"""
Return the semgent name associated with the data update.
:returns: segment name
:rtype: str
"""
return
self
.
_segment_name
def
__str__
(
self
):
"""Return string representation."""
return
"SegmentChange - changeNumber=%d, name=%s"
%
(
self
.
change_number
,
self
.
segment_name
)
class
RBSChangeUpdate
(
BaseUpdate
):
"""rbs Change notification."""
def
__init__
(
self
,
channel
,
timestamp
,
change_number
,
previous_change_number
,
rbs_definition
,
compression
):
"""Class constructor."""
BaseUpdate
.
__init__
(
self
,
channel
,
timestamp
,
change_number
)
self
.
_previous_change_number
=
previous_change_number
self
.
_object_definition
=
rbs_definition
self
.
_compression
=
compression
@
property
def
update_type
(
self
):
# pylint:disable=no-self-use
"""
Return the message type.
:returns: The type of this parsed Update.
:rtype: UpdateType
"""
return
UpdateType
.
RB_SEGMENT_UPDATE
@
property
def
previous_change_number
(
self
):
# pylint:disable=no-self-use
"""
Return previous change number
:returns: The previous change number
:rtype: int
"""
return
self
.
_previous_change_number
@
property
def
object_definition
(
self
):
# pylint:disable=no-self-use
"""
Return rbs definition
:returns: The new rbs definition
:rtype: str
"""
return
self
.
_object_definition
@
property
def
compression
(
self
):
# pylint:disable=no-self-use
"""
Return previous compression type
:returns: The compression type
:rtype: int
"""
return
self
.
_compression
def
__str__
(
self
):
"""Return string representation."""
return
"RBSChange - changeNumber=%d"
%
(
self
.
change_number
)
class
ControlMessage
(
BaseMessage
):
"""Control notification."""
def
__init__
(
self
,
channel
,
timestamp
,
control_type
):
"""Class constructor."""
BaseMessage
.
__init__
(
self
,
channel
,
timestamp
)
self
.
_control_type
=
ControlType
(
control_type
)
@
property
def
message_type
(
self
):
# pylint:disable=no-self-use
"""
Return the message type.
:returns: The type of this parsed event.
:rtype: MessageType
"""
return
MessageType
.
CONTROL
@
property
def
control_type
(
self
):
"""
Return the associated control type.
:returns: control type
:rtype: ControlType
"""
return
self
.
_control_type
def
__str__
(
self
):
"""Return string representation."""
return
"Control - type=%s"
%
(
self
.
control_type
.
name
)
def
_parse_update
(
channel
,
timestamp
,
data
):
"""
Parse a message of update type.
:param channel: channel name
:type data: str
:param data: raw incoming event
:type data: dict
:returns: Parsed ably error notification.
:rtype: BaseUpdate
"""
update_type
=
UpdateType
(
data
[
'type'
])
change_number
=
data
[
'changeNumber'
]
if
update_type
==
UpdateType
.
SPLIT_UPDATE
and
change_number
is
not
None
:
return
SplitChangeUpdate
(
channel
,
timestamp
,
change_number
,
data
.
get
(
'pcn'
),
data
.
get
(
'd'
),
data
.
get
(
'c'
))
if
update_type
==
UpdateType
.
RB_SEGMENT_UPDATE
and
change_number
is
not
None
:
return
RBSChangeUpdate
(
channel
,
timestamp
,
change_number
,
data
.
get
(
'pcn'
),
data
.
get
(
'd'
),
data
.
get
(
'c'
))
elif
update_type
==
UpdateType
.
SPLIT_KILL
and
change_number
is
not
None
:
return
SplitKillUpdate
(
channel
,
timestamp
,
change_number
,
data
[
'splitName'
],
data
[
'defaultTreatment'
])
elif
update_type
==
UpdateType
.
SEGMENT_UPDATE
:
return
SegmentChangeUpdate
(
channel
,
timestamp
,
change_number
,
data
[
'segmentName'
])
raise
EventParsingException
(
'unrecognized event type %s'
%
update_type
)
def
_parse_message
(
data
):
"""
Parse a message event into a concrete class.
:param data: raw incoming event.
:type data: dict:
:returns: Parsed ably error notification.
:rtype: BaseEvent
"""
if
not
all
(
k
in
data
for
k
in
[
'data'
,
'channel'
]):
return
None
channel
=
data
[
'channel'
]
timestamp
=
data
[
'timestamp'
]
parsed_data
=
json
.
loads
(
data
[
'data'
])
if
data
.
get
(
'name'
)
==
TAG_OCCUPANCY
:
return
OccupancyMessage
(
channel
,
timestamp
,
parsed_data
[
'metrics'
][
'publishers'
])
elif
parsed_data
[
'type'
]
==
'CONTROL'
:
return
ControlMessage
(
channel
,
timestamp
,
parsed_data
[
'controlType'
])
elif
parsed_data
[
'type'
]
in
UpdateType
.
__members__
:
return
_parse_update
(
channel
,
timestamp
,
parsed_data
)
raise
EventParsingException
(
'unrecognized message type %s'
%
parsed_data
[
'type'
])
def
_parse_error
(
data
):
"""
Parse an error message into a concrete class.
:param data: raw incoming event.
:type data: dict:
:returns: Parsed ably error notification.
:rtype: AblyError
"""
return
AblyError
(
data
.
get
(
'code'
),
data
.
get
(
'statusCode'
),
data
.
get
(
'message'
),
data
.
get
(
'href'
))
def
parse_incoming_event
(
raw_event
):
"""
Parse a raw event as received by the sse client.
:param raw_event: raw SSE Event
:type raw_event: splitio.push.sse.SSEEvent
:returns: an event parsed to it's concrete type.
:rtype: BaseEvent
"""
if
raw_event
is
None
:
return
None
try
:
parsed_data
=
json
.
loads
(
raw_event
.
data
)
except
Exception
as
exc
:
# pylint:disable=broad-except
raise
EventParsingException
(
'Error parsing json'
)
from
exc
try
:
event_type
=
EventType
(
raw_event
.
event
)
except
ValueError
as
exc
:
raise
Exception
(
'unknown event type %s'
%
raw_event
.
event
)
from
exc
return
{
EventType
.
ERROR
:
_parse_error
,
EventType
.
MESSAGE
:
_parse_message
,
}[
event_type
](
parsed_data
)
Back
|
FazBrowse Home
|
New Git URL