added fetch segment from sdk endpoint

README.md

@@ -281,6 +281,39 @@ definition= {"treatments":[ {"name":"on"},{"name":"off"}],
 }
 splitDef.submit_change_request(definition, 'UPDATE', 'updating default rule', 'comment', ['[email protected]'], '')
 ```
+### Segments Keys
+This build allows fetching segments keys from SDK Endpoints to speedup the download for big sized segments. 
+
+The function `client.segment_definitions.get_all_keys` takes 2 parameters, the segment name and an Environment object retrieved from `client.environments`
+If the function is successful, it will return a json below with set of all segment keys and the total keys count, see example below:
+```json
+{
+    "keys": {"key1", "key2", "key3"},
+    "count": 3
+}
+```
+If any network issue or http returned codes are not within 200-300 range, None is returned. All errors are logged in debug mode. 
+
+Below an example of fetching all segments keys in en environment.
+
+```python
+ws = client.workspaces.find("Default")
+env = client.environments.find("Production", ws.id)
+env.sdkApiToken = "SDK API Key (Server side)"
+
+for segDef in client.segment_definitions.list(env.id, ws.id):
+    print(segDef.name)
+    print("============")
+    keys = client.segment_definitions.get_all_keys(segDef.name, env)
+    if keys == None:
+        print("Failed to get keys, check debug logs")
+    else:
+        print ("Segment Keys: ")
+        print(keys)
+    print("\n")
+
+print("done.")
+```
 
 ### Rule-Based Segments
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "splitapiclient"
-version = "3.5.4"
+version = "3.6.0.rc.1"
 description = "This Python Library provide full support for Split REST Admin API, allow creating, deleting and editing Environments, Splits, Split Definitions, Segments, Segment Keys, Users, Groups, API Keys, Change Requests, Attributes and Identities"
 classifiers = [
     "Programming Language :: Python :: 3",

splitapiclient/microclients/segment_definition_microclient.py

@@ -3,6 +3,19 @@
     UnknownApiClientError
 from splitapiclient.util.logger import LOGGER
 from splitapiclient.util.helpers import as_dict
+from splitapiclient.util.fetch_options import FetchOptions, Backoff, build_fetch
+from splitapiclient.resources import Environment
+from splitapiclient.resources import segments
+from splitapiclient.util.logger import LOGGER
+
+import requests
+import json
+import time
+
+_ON_DEMAND_FETCH_BACKOFF_BASE = 10  # backoff base starting at 10 seconds
+_ON_DEMAND_FETCH_BACKOFF_MAX_WAIT = 60  # don't sleep for more than 1 minute
+_ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES = 10
+SDK_URL = 'https://sdk.split.io/api'
 
 class SegmentDefinitionMicroClient:
     '''
@@ -188,3 +201,206 @@ def remove_keys(self, segment_name, environment_id, data):
         )
         return True
 
+
+    def get_all_keys(self, segment_name, environment):
+        '''
+        Get list of keys in segment in environment
+
+        :param data: None
+        :param apiclient: If this instance wasn't returned by the client,
+            the IdentifyClient instance should be passed in order to perform the
+            http call
+
+        :returns: string of keys instance
+        :rtype: string
+        '''
+        if not self._validate_sdkapi_key(environment.sdkApiToken):
+            return None
+
+        self._name = segment_name
+        self._sdk_api_key = environment.sdkApiToken
+        self._segment_storage = None
+        self._segment_change_number = None
+        self._metadata =  {
+            'SplitSDKVersion': 'python-3.6.0-wrapper',
+        }
+        self._backoff = Backoff(
+                                _ON_DEMAND_FETCH_BACKOFF_BASE,
+                                _ON_DEMAND_FETCH_BACKOFF_MAX_WAIT)
+
+        if self._get_segment_from_sdk_endpoint(self._name):
+            keys = self._segment_storage.keys
+            self._segment_storage = None
+            return {
+                "keys": keys,
+                "count": len(keys)
+            }
+
+        LOGGER.error("Failed to fetch segment %s keys", self._name)
+        return None
+
+    def _validate_sdkapi_key(self, sdkApiToken):
+        if sdkApiToken == None:
+            LOGGER.error("Environment object does not have the SDK Api Key set, please set it before calling this method.")
+            return False
+
+        if not isinstance(sdkApiToken, str):
+            LOGGER.error("SDK Api Key must be a string, please use a string to set it before calling this method.")
+            return False
+
+        if len(sdkApiToken) != 36:
+            LOGGER.error("SDK Api Key string is invalid, please set it before calling this method.")
+            return False
+
+        return True
+
+    def _fetch_until(self, segment_name, fetch_options, till=None):
+        """
+        Hit endpoint, update storage and return when since==till.
+
+        :param segment_name: Name of the segment to update.
+        :type segment_name: str
+
+        :param fetch_options Fetch options for getting segment definitions.
+        :type fetch_options splitio.api.FetchOptions
+
+        :param till: Passed till from Streaming.
+        :type till: int
+
+        :return: last change number
+        :rtype: int
+        """
+        while True:  # Fetch until since==till
+            change_number = self._segment_change_number
+            if change_number is None:
+                change_number = -1
+            if till is not None and till < change_number:
+                # the passed till is less than change_number, no need to perform updates
+                return change_number
+
+            try:
+                segment_changes = self._fetch_segment_api(segment_name, change_number,
+                                                          fetch_options)
+                if segment_changes == None:
+                    return None
+
+            except Exception as exc:
+                LOGGER.debug('Exception raised while fetching segment %s', segment_name)
+                LOGGER.error('Exception information: %s', str(exc))
+                return None
+
+            if change_number == -1:  # first time fetching the segment
+                new_segment = segments.from_raw(segment_changes)
+                self._segment_storage = new_segment
+                self._segment_change_number = new_segment.change_number
+            else:
+                self._segment_change_number = segment_changes['till']
+                self._segment_storage.keys.update(segment_changes['added'])
+                [self._segment_storage.keys.remove(key) for key in  segment_changes['removed']]
+
+            if segment_changes['till'] == segment_changes['since']:
+                return segment_changes['till']
+
+    def _attempt_segment_sync(self, segment_name, fetch_options, till=None):
+        """
+        Hit endpoint, update storage and return True if sync is complete.
+
+        :param segment_name: Name of the segment to update.
+        :type segment_name: str
+
+        :param fetch_options Fetch options for getting feature flag definitions.
+        :type fetch_options splitio.api.FetchOptions
+
+        :param till: Passed till from Streaming.
+        :type till: int
+
+        :return: Flags to check if it should perform bypass or operation ended
+        :rtype: bool, int, int
+        """
+        self._backoff.reset()
+        remaining_attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES
+        while True:
+            remaining_attempts -= 1
+            change_number = self._fetch_until(segment_name, fetch_options, till)
+            if change_number == None:
+                return False, 0, None
+
+            if till is None or till <= change_number:
+                return True, remaining_attempts, change_number
+
+            elif remaining_attempts <= 0:
+                return False, remaining_attempts, change_number
+
+            how_long = self._backoff.get()
+            time.sleep(how_long)
+
+    def _get_segment_from_sdk_endpoint(self, segment_name, till=None):
+        """
+        Update a segment from queue
+
+        :param segment_name: Name of the segment to update.
+        :type segment_name: str
+
+        :param till: ChangeNumber received.
+        :type till: int
+
+        :return: True if no error occurs. False otherwise.
+        :rtype: bool
+        """
+        fetch_options = FetchOptions(True)  # Set Cache-Control to no-cache
+        successful_sync, remaining_attempts, change_number = self._attempt_segment_sync(segment_name, fetch_options, till)
+        if change_number == None:
+            return False
+
+        attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - remaining_attempts
+        if successful_sync:  # succedeed sync
+            LOGGER.debug('Refresh completed in %d attempts.', attempts)
+            return True
+        with_cdn_bypass = FetchOptions(True, change_number)  # Set flag for bypassing CDN
+        without_cdn_successful_sync, remaining_attempts, change_number = self._attempt_segment_sync(segment_name, with_cdn_bypass, till)
+        if change_number == None:
+            return False
+
+        without_cdn_attempts = _ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - remaining_attempts
+        if without_cdn_successful_sync:
+            LOGGER.debug('Refresh completed bypassing the CDN in %d attempts.',
+                          without_cdn_attempts)
+            return True
+
+        LOGGER.debug('No changes fetched after %d attempts with CDN bypassed.',
+                        without_cdn_attempts)
+        return False
+
+    def _fetch_segment_api(self, segment_name, change_number, fetch_options):
+        try:
+            query, extra_headers = build_fetch(change_number, fetch_options, self._metadata)
+            response = requests.get(
+                SDK_URL + '/segmentChanges/{segment_name}'.format(segment_name=segment_name),
+                headers=self._build_basic_headers(extra_headers),
+                params=query,
+            )
+            if 200 <= response.status_code < 300:
+                return json.loads(response.text)
+
+            return None
+        except Exception as exc:
+            LOGGER.debug(
+                'Error fetching %s because an exception was raised by the HTTPClient',
+                segment_name)
+            LOGGER.error(str(exc))
+            return None
+
+    def _build_basic_headers(self, extra_headers):
+        """
+        Build basic headers with auth.
+
+        :param sdk_key: API token used to identify backend calls.
+        :type sdk_key: str
+        """
+        headers = {
+            'Content-Type': 'application/json',
+            'Authorization': "Bearer %s" % self._sdk_api_key
+        }
+        if extra_headers is not None:
+            headers.update(extra_headers)
+        return headers

splitapiclient/resources/environment.py

@@ -52,7 +52,7 @@ class Environment(BaseResource):
         "status" : "string"
     }
 
-    def __init__(self, data=None, workspace_id=None, client=None):
+    def __init__(self, data=None, workspace_id=None, client=None, sdk_apikey=None):
         '''
         '''
         if not data:
@@ -70,7 +70,16 @@ def __init__(self, data=None, workspace_id=None, client=None):
         self._changePermissions = data.get("changePermissions") if "changePermissions" in data else {}
         self._apiTokens = data.get("apiTokens") if "apiTokens" in data else {}
         self._client = client
+        self._sdk_apikey = sdk_apikey
 
+    @property
+    def sdkApiToken(self):
+        return self._sdk_apikey
+
+    @sdkApiToken.setter
+    def sdkApiToken(self, value):
+        self._sdk_apikey = value
+
     @property
     def apiTokens(self):
         return self._apiTokens

splitapiclient/resources/segment_definition.py

@@ -3,7 +3,7 @@
 from splitapiclient.resources.base_resource import BaseResource
 from splitapiclient.util.helpers import require_client, as_dict
 from splitapiclient.resources import TrafficType
-from splitapiclient.resources import Environment
+
 import csv
 
 class SegmentDefinition(BaseResource):
@@ -172,3 +172,4 @@ def submit_change_request(self, keys, operation_type, title, comment, approvers,
             data['rolloutStatus'] = {'id': rollout_status_id}
         imc = require_client('ChangeRequest', self._client, apiclient)
         return imc.submit_change_request(self._environment['id'], workspace_id, data)
+