feat: added switcher.default_result() fall back feature

README.md

@@ -209,9 +209,15 @@ if switcher.is_on():
 Chain multiple validation strategies for comprehensive feature control:
 
 ```python
-# Validate user, network, and other criteria in one call
-is_enabled = switcher.check_value('premium_user') \
+is_enabled = switcher \
+                    # VALUE strategy
+                    .check_value('premium_user') \
+                    # NETWORK strategy
                     .check_network('192.168.1.0/24') \
+                    # Fallback value if API fails
+                    .default_result(True) \
+                    # Cache result for 1 second
+                    .throttle(1000) \
                     .is_on('PREMIUM_FEATURES')
 
 if is_enabled:

switcher_client/lib/types.py

@@ -6,6 +6,10 @@ def __init__(self, result: bool, reason: Optional[str], metadata: Optional[dict]
         self.reason = reason
         self.metadata = metadata
 
+    @staticmethod
+    def create(result: bool, reason: Optional[str], metadata: Optional[dict] = None) -> 'ResultDetail':
+        return ResultDetail(result=result, reason=reason, metadata=metadata)
+
     @staticmethod
     def disabled(reason: str, metadata: Optional[dict] = None) -> 'ResultDetail':
         return ResultDetail(result=False, reason=reason, metadata=metadata)

switcher_client/switcher.py

@@ -90,18 +90,36 @@ def _notify_error(self, error: Exception):
 
     def _execute_remote_criteria(self):
         """ Execute remote criteria """
-        token = GlobalAuth.get_token()
-        response = Remote.check_criteria(token, self._context, self)
+        try:
+            token = GlobalAuth.get_token()
+            response = Remote.check_criteria(token, self._context, self)
 
-        if self._can_log():
-            ExecutionLogger.add(response, self._key, self._input)
+            if self._can_log():
+                ExecutionLogger.add(response, self._key, self._input)
 
-        return response
+            return response
+        except Exception as e:
+            return self._get_default_result_or_raise(e)
 
     def _execute_local_criteria(self):
         """ Execute local criteria """
-        return Resolver.check_criteria(GlobalSnapshot.snapshot(), self)
+        try:
+            response = Resolver.check_criteria(GlobalSnapshot.snapshot(), self)
+            if self._can_log():
+                ExecutionLogger.add(response, self._key, self._input)
+
+            return response
+        except Exception as e:
+            return self._get_default_result_or_raise(e)
 
     def _can_log(self) -> bool:
         """ Check if logging is enabled """
-        return self._context.options.logger and self._key is not None
+        return self._context.options.logger and self._key is not None
+
+    def _get_default_result_or_raise(self, e) -> ResultDetail:
+        """ Get default result if set, otherwise raise the error """
+        if self._default_result is None:
+            raise e
+
+        self._notify_error(e)
+        return ResultDetail.create(result=self._default_result, reason="Default result")

switcher_client/switcher_data.py

@@ -11,6 +11,7 @@ def __init__(self, key: Optional[str] = None):
         self._show_details = False
         self._throttle_period = 0
         self._next_refresh_time = 0 # timestamp
+        self._default_result = None
 
     def check(self, strategy_type: str, input: str)-> Self:
         """ Adds a strategy for validation """
@@ -47,4 +48,9 @@ def throttle(self, period: int) -> Self:
         if self._next_refresh_time == 0:
             self._next_refresh_time = int(datetime.now().timestamp() * 1000) + period
 
+        return self
+
+    def default_result(self, result: bool) -> Self:
+        """ Sets the default result for the switcher """
+        self._default_result = result
         return self

tests/test_switcher_local.py

@@ -4,6 +4,8 @@
 from switcher_client.errors import LocalCriteriaError
 from switcher_client.lib.utils.timed_match.timed_match import TimedMatch
 
+async_error = None
+
 def test_local():
     """ Should use local Snapshot to evaluate the switcher """
 
@@ -156,6 +158,25 @@ def test_local_no_key_found():
     except LocalCriteriaError as e:
         assert str(e) == "Config with key 'INVALID_KEY' not found in the snapshot"
 
+def test_local_err_check_criteria_default_result():
+    """ Should return default result when check criteria fails """
+
+    # given
+    given_context('tests/snapshots')
+    snapshot_version = Client.load_snapshot()
+
+    globals().update(async_error=None)
+    Client.subscribe_notify_error(lambda error: globals().update(async_error=str(error)))
+    switcher = Client.get_switcher()
+
+    # test
+    assert snapshot_version == 1
+
+    feature = switcher.default_result(True).is_on_with_details('INVALID_KEY')
+    assert feature.result is True
+    assert feature.reason == 'Default result'
+    assert async_error == "Config with key 'INVALID_KEY' not found in the snapshot"
+
 def test_local_no_snapshot():
     """ Should raise an error when no snapshot is loaded """
 
@@ -177,6 +198,7 @@ def given_context(snapshot_location: str, environment: str = 'default') -> None:
         environment=environment,
         options=ContextOptions(
             local=True,
+            logger=True,
             snapshot_location=snapshot_location
         )
     )

tests/test_switcher_remote.py

@@ -8,6 +8,8 @@
 from switcher_client import Client
 from switcher_client.lib.globals.global_auth import GlobalAuth
 
+async_error = None
+
 def test_remote(httpx_mock):
     """ Should call the remote API with success """
 
@@ -163,6 +165,24 @@ def test_remote_err_check_criteria(httpx_mock):
 
     assert '[check_criteria] failed with status: 500' in str(excinfo.value)
 
+def test_remote_err_check_criteria_default_result(httpx_mock):
+    """ Should return the default result when the check criteria fails """
+
+    # given
+    given_auth(httpx_mock)
+    given_check_criteria(httpx_mock, show_details=True, status=500)
+    given_context()
+
+    globals().update(async_error=None)
+    Client.subscribe_notify_error(lambda error: globals().update(async_error=str(error)))
+    switcher = Client.get_switcher()
+
+    # test
+    feature = switcher.default_result(True).is_on_with_details('MY_SWITCHER')
+    assert feature.result is True
+    assert feature.reason == 'Default result'
+    assert async_error == '[check_criteria] failed with status: 500'
+
 # Helpers
 
 def given_context(url='https://api.switcherapi.com', api_key='[API_KEY]'):