FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

Fme 12223 sdk events segments by chillaq · Pull Request #613 · splitio/python-client · GitHub

/ python-client Public

Fme 12223 sdk events segments #613

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chillaq merged 1 commit into from
Jan 14, 2026
Conversation 1 Commits 1 Checks 2 Files changed
Merged

Fme 12223 sdk events segments #613

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension .py  (12) All 1 file type selected
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
8 changes: 4 additions & 4 deletions splitio/client/factory.py
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -551,8 +551,8 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None, # pyl
events_queue = queue.Queue()
storages = {
'splits': InMemorySplitStorage(events_queue, cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
'segments': InMemorySegmentStorage(),
'rule_based_segments': InMemoryRuleBasedSegmentStorage(),
'segments': InMemorySegmentStorage(events_queue),
'rule_based_segments': InMemoryRuleBasedSegmentStorage(events_queue),
'impressions': InMemoryImpressionStorage(cfg['impressionsQueueSize'], telemetry_runtime_producer),
'events': InMemoryEventStorage(cfg['eventsQueueSize'], telemetry_runtime_producer),
}
Expand Down Expand Up @@ -1101,8 +1101,8 @@ def _build_localhost_factory(cfg):
events_queue = queue.Queue()
storages = {
'splits': InMemorySplitStorage(events_queue, cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
'segments': InMemorySegmentStorage(), # not used, just to avoid possible future errors.
'rule_based_segments': InMemoryRuleBasedSegmentStorage(),
'segments': InMemorySegmentStorage(events_queue), # not used, just to avoid possible future errors.
'rule_based_segments': InMemoryRuleBasedSegmentStorage(events_queue),
'impressions': LocalhostImpressionsStorage(),
'events': LocalhostEventsStorage(),
}
Expand Down
24 changes: 20 additions & 4 deletions splitio/storage/inmemmory.py
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -113,11 +113,12 @@ def remove_flag_set(self, flag_sets, feature_flag_name, should_filter):
class InMemoryRuleBasedSegmentStorage(RuleBasedSegmentsStorage):
"""InMemory implementation of a feature flag storage base."""

def __init__(self):
def __init__(self, internal_event_queue):
"""Constructor."""
self._lock = threading.RLock()
self._rule_based_segments = {}
self._change_number = -1
self._internal_event_queue = internal_event_queue

def clear(self):
"""
Expand Down Expand Up @@ -153,6 +154,10 @@ def update(self, to_add, to_delete, new_change_number):
[self._put(add_segment) for add_segment in to_add]
[self._remove(delete_segment) for delete_segment in to_delete]
self._set_change_number(new_change_number)
self._internal_event_queue.put(
SdkInternalEventNotification(
SdkInternalEvent.RB_SEGMENTS_UPDATED,
EventsMetadata(SdkEventType.SEGMENT_UPDATE, {})))

def _put(self, rule_based_segment):
"""
Expand Down Expand Up @@ -934,11 +939,12 @@ async def is_flag_set_exist(self, flag_set):
class InMemorySegmentStorage(SegmentStorage):
"""In-memory implementation of a segment storage."""

def __init__(self):
def __init__(self, internal_event_queue):
"""Constructor."""
self._segments = {}
self._change_numbers = {}
self._lock = threading.RLock()
self._internal_event_queue = internal_event_queue

def get(self, segment_name):
"""
Expand Down Expand Up @@ -968,9 +974,14 @@ def put(self, segment):
with self._lock:
self._segments[segment.name] = segment

self._internal_event_queue.put(
SdkInternalEventNotification(
SdkInternalEvent.SEGMENTS_UPDATED,
EventsMetadata(SdkEventType.SEGMENT_UPDATE, {})))

def update(self, segment_name, to_add, to_remove, change_number=None):
"""
Update a feature flag. Create it if it doesn't exist.
Update a segment. Create it if it doesn't exist.

:param segment_name: Name of the segment to update.
:type segment_name: str
Expand All @@ -988,6 +999,11 @@ def update(self, segment_name, to_add, to_remove, change_number=None):
if change_number is not None:
self._segments[segment_name].change_number = change_number

self._internal_event_queue.put(
SdkInternalEventNotification(
SdkInternalEvent.SEGMENTS_UPDATED,
EventsMetadata(SdkEventType.SEGMENT_UPDATE, {})))

def get_change_number(self, segment_name):
"""
Retrieve latest change number for a segment.
Expand Down Expand Up @@ -1100,7 +1116,7 @@ async def put(self, segment):

async def update(self, segment_name, to_add, to_remove, change_number=None):
"""
Update a feature flag. Create it if it doesn't exist.
Update a segment. Create it if it doesn't exist.

:param segment_name: Name of the segment to update.
:type segment_name: str
Expand Down
62 changes: 31 additions & 31 deletions tests/client/test_client.py
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -39,8 +39,8 @@ def test_get_treatment(self, mocker):
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
events_queue = queue.Queue()
split_storage = InMemorySplitStorage(events_queue)
segment_storage = InMemorySegmentStorage()
rb_segment_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage(events_queue)
rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer)
event_storage = mocker.Mock(spec=EventStorage)
Expand Down Expand Up @@ -117,8 +117,8 @@ def test_get_treatment_with_config(self, mocker):
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
events_queue = queue.Queue()
split_storage = InMemorySplitStorage(events_queue)
segment_storage = InMemorySegmentStorage()
rb_segment_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage(events_queue)
rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer)
impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer)
Expand Down Expand Up @@ -195,8 +195,8 @@ def test_get_treatments(self, mocker):
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
events_queue = queue.Queue()
split_storage = InMemorySplitStorage(events_queue)
segment_storage = InMemorySegmentStorage()
rb_segment_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage(events_queue)
rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer)
impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer)
Expand Down Expand Up @@ -276,8 +276,8 @@ def test_get_treatments_by_flag_set(self, mocker):
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
events_queue = queue.Queue()
split_storage = InMemorySplitStorage(events_queue)
segment_storage = InMemorySegmentStorage()
rb_segment_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage(events_queue)
rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer)
impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer)
Expand Down Expand Up @@ -356,8 +356,8 @@ def test_get_treatments_by_flag_sets(self, mocker):
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
events_queue = queue.Queue()
split_storage = InMemorySplitStorage(events_queue)
segment_storage = InMemorySegmentStorage()
rb_segment_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage(events_queue)
rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer)
impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer)
Expand Down Expand Up @@ -436,8 +436,8 @@ def test_get_treatments_with_config(self, mocker):
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
events_queue = queue.Queue()
split_storage = InMemorySplitStorage(events_queue)
segment_storage = InMemorySegmentStorage()
rb_segment_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage(events_queue)
rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer)
impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer)
Expand Down Expand Up @@ -520,8 +520,8 @@ def test_get_treatments_with_config_by_flag_set(self, mocker):
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
events_queue = queue.Queue()
split_storage = InMemorySplitStorage(events_queue)
segment_storage = InMemorySegmentStorage()
rb_segment_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage(events_queue)
rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer)
impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer)
Expand Down Expand Up @@ -601,8 +601,8 @@ def test_get_treatments_with_config_by_flag_sets(self, mocker):
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
events_queue = queue.Queue()
split_storage = InMemorySplitStorage(events_queue)
segment_storage = InMemorySegmentStorage()
rb_segment_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage(events_queue)
rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer)
impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer)
Expand Down Expand Up @@ -682,8 +682,8 @@ def test_impression_toggle_optimized(self, mocker):
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
events_queue = queue.Queue()
split_storage = InMemorySplitStorage(events_queue)
segment_storage = InMemorySegmentStorage()
rb_segment_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage(events_queue)
rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer)
event_storage = mocker.Mock(spec=EventStorage)
Expand Down Expand Up @@ -747,8 +747,8 @@ def test_impression_toggle_debug(self, mocker):
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
events_queue = queue.Queue()
split_storage = InMemorySplitStorage(events_queue)
segment_storage = InMemorySegmentStorage()
rb_segment_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage(events_queue)
rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer)
event_storage = mocker.Mock(spec=EventStorage)
Expand Down Expand Up @@ -812,8 +812,8 @@ def test_impression_toggle_none(self, mocker):
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
events_queue = queue.Queue()
split_storage = InMemorySplitStorage(events_queue)
segment_storage = InMemorySegmentStorage()
rb_segment_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage(events_queue)
rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer)
event_storage = mocker.Mock(spec=EventStorage)
Expand Down Expand Up @@ -953,8 +953,8 @@ def test_evaluations_before_running_post_fork(self, mocker):
impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer)
events_queue = queue.Queue()
split_storage = InMemorySplitStorage(events_queue)
segment_storage = InMemorySegmentStorage()
rb_segment_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage(events_queue)
rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0])], [], -1)
destroyed_property = mocker.PropertyMock()
destroyed_property.return_value = False
Expand Down Expand Up @@ -1035,8 +1035,8 @@ def test_telemetry_not_ready(self, mocker):
impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer)
events_queue = queue.Queue()
split_storage = InMemorySplitStorage(events_queue)
segment_storage = InMemorySegmentStorage()
rb_segment_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage(events_queue)
rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0])], [], -1)
recorder = StandardRecorder(impmanager, mocker.Mock(), mocker.Mock(), telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer())
factory = SplitFactory('localhost',
Expand Down Expand Up @@ -1071,7 +1071,7 @@ def test_telemetry_record_treatment_exception(self, mocker):
split_storage = InMemorySplitStorage(events_queue)
split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0])], [], -1)
segment_storage = mocker.Mock(spec=SegmentStorage)
rb_segment_storage = InMemoryRuleBasedSegmentStorage()
rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
impression_storage = mocker.Mock(spec=ImpressionStorage)
event_storage = mocker.Mock(spec=EventStorage)
destroyed_property = mocker.PropertyMock()
Expand Down Expand Up @@ -1175,8 +1175,8 @@ def test_telemetry_method_latency(self, mocker):
impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer)
events_queue = queue.Queue()
split_storage = InMemorySplitStorage(events_queue)
segment_storage = InMemorySegmentStorage()
rb_segment_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage(events_queue)
rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0])], [], -1)
recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer())
destroyed_property = mocker.PropertyMock()
Expand Down Expand Up @@ -1288,8 +1288,8 @@ def test_impressions_properties(self, mocker):
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
events_queue = queue.Queue()
split_storage = InMemorySplitStorage(events_queue)
segment_storage = InMemorySegmentStorage()
rb_segment_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage(events_queue)
rb_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer)
event_storage = mocker.Mock(spec=EventStorage)
Expand Down
20 changes: 10 additions & 10 deletions tests/engine/test_evaluator.py
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -264,8 +264,8 @@ def test_evaluate_treatment_with_rbs_in_condition(self):
e = evaluator.Evaluator(splitters.Splitter())
events_queue = queue.Queue()
splits_storage = InMemorySplitStorage(events_queue)
rbs_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage()
rbs_storage = InMemoryRuleBasedSegmentStorage(events_queue)
segment_storage = InMemorySegmentStorage(events_queue)
evaluation_facctory = EvaluationDataFactory(splits_storage, segment_storage, rbs_storage)

rbs_segments = os.path.join(os.path.dirname(__file__), 'files', 'rule_base_segments.json')
Expand All @@ -291,8 +291,8 @@ def test_using_segment_in_excluded(self):
e = evaluator.Evaluator(splitters.Splitter())
events_queue = queue.Queue()
splits_storage = InMemorySplitStorage(events_queue)
rbs_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage()
rbs_storage = InMemoryRuleBasedSegmentStorage(events_queue)
segment_storage = InMemorySegmentStorage(events_queue)
evaluation_facctory = EvaluationDataFactory(splits_storage, segment_storage, rbs_storage)

mocked_split = Split('some', 12345, False, 'off', 'user', Status.ACTIVE, 12, split_conditions, 1.2, 100, 1234, {}, None, False, [])
Expand All @@ -316,8 +316,8 @@ def test_using_rbs_in_excluded(self):
e = evaluator.Evaluator(splitters.Splitter())
events_queue = queue.Queue()
splits_storage = InMemorySplitStorage(events_queue)
rbs_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage()
rbs_storage = InMemoryRuleBasedSegmentStorage(events_queue)
segment_storage = InMemorySegmentStorage(events_queue)
evaluation_facctory = EvaluationDataFactory(splits_storage, segment_storage, rbs_storage)

mocked_split = Split('some', 12345, False, 'off', 'user', Status.ACTIVE, 12, split_conditions, 1.2, 100, 1234, {}, None, False, [])
Expand All @@ -340,8 +340,8 @@ def test_prerequisites(self):
e = evaluator.Evaluator(splitters.Splitter())
events_queue = queue.Queue()
splits_storage = InMemorySplitStorage(events_queue)
rbs_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage()
rbs_storage = InMemoryRuleBasedSegmentStorage(events_queue)
segment_storage = InMemorySegmentStorage(events_queue)
evaluation_facctory = EvaluationDataFactory(splits_storage, segment_storage, rbs_storage)

rbs = rule_based_segments.from_raw(data["rbs"]["d"][0])
Expand Down Expand Up @@ -549,8 +549,8 @@ def test_get_context(self):
split2 = Split('split2', 12345, False, 'off', 'user', Status.ACTIVE, 12, split_conditions, 1.2, 100, 1234, {}, None, False, [])
events_queue = queue.Queue()
flag_storage = InMemorySplitStorage(events_queue, [])
segment_storage = InMemorySegmentStorage()
rbs_segment_storage = InMemoryRuleBasedSegmentStorage()
segment_storage = InMemorySegmentStorage(events_queue)
rbs_segment_storage = InMemoryRuleBasedSegmentStorage(events_queue)
flag_storage.update([mocked_split, split2], [], -1)
rbs = copy.deepcopy(rbs_raw)
rbs['conditions'].append(
Expand Down
Loading
Back | FazBrowse Home | New Git URL