Initial implementation of offline mode (single client class)

Author: matthewelwell

flagsmith/flagsmith.py

@@ -14,6 +14,7 @@
 from flagsmith.analytics import AnalyticsProcessor
 from flagsmith.exceptions import FlagsmithAPIError, FlagsmithClientError
 from flagsmith.models import DefaultFlag, Flags, Segment
+from flagsmith.offline_handlers import BaseOfflineModeHandler
 from flagsmith.polling_manager import EnvironmentDataPollingManager
 from flagsmith.utils.identities import generate_identities_data
 
@@ -39,8 +40,8 @@ class Flagsmith:
 
     def __init__(
         self,
-        environment_key: str,
-        api_url: str = DEFAULT_API_URL,
+        environment_key: str = None,
+        api_url: str = None,
         custom_headers: typing.Dict[str, typing.Any] = None,
         request_timeout_seconds: int = None,
         enable_local_evaluation: bool = False,
@@ -49,6 +50,8 @@ def __init__(
         enable_analytics: bool = False,
         default_flag_handler: typing.Callable[[str], DefaultFlag] = None,
         proxies: typing.Dict[str, str] = None,
+        offline_mode: bool = True,
+        offline_handler: BaseOfflineModeHandler = None,
     ):
         """
         :param environment_key: The environment key obtained from Flagsmith interface
@@ -69,37 +72,45 @@ def __init__(
             requested
         :param proxies: as per https://requests.readthedocs.io/en/latest/api/#requests.Session.proxies
         """
-        self.session = requests.Session()
-        self.session.headers.update(
-            **{"X-Environment-Key": environment_key}, **(custom_headers or {})
-        )
-        self.session.proxies.update(proxies or {})
-        retries = retries or Retry(total=3, backoff_factor=0.1)
-
-        self.api_url = api_url if api_url.endswith("/") else f"{api_url}/"
-        self.request_timeout_seconds = request_timeout_seconds
-        self.session.mount(self.api_url, HTTPAdapter(max_retries=retries))
-
-        self.environment_flags_url = f"{self.api_url}flags/"
-        self.identities_url = f"{self.api_url}identities/"
-        self.environment_url = f"{self.api_url}environment-document/"
-
-        self._environment = None
-        if enable_local_evaluation:
-            if not environment_key.startswith("ser."):
-                raise ValueError(
-                    "In order to use local evaluation, please generate a server key "
-                    "in the environment settings page."
-                )
-
-            self.environment_data_polling_manager_thread = (
-                EnvironmentDataPollingManager(
-                    main=self,
-                    refresh_interval_seconds=environment_refresh_interval_seconds,
-                    daemon=True,  # noqa
-                )
+        if offline_mode:
+            self.offline_mode = True
+            self.offline_handler = offline_handler
+            self.update_environment()
+        elif not (environment_key and api_url):
+            raise ValueError("environment_key and api_url are required.")
+        else:
+            self.offline_mode = False
+            self.session = requests.Session()
+            self.session.headers.update(
+                **{"X-Environment-Key": environment_key}, **(custom_headers or {})
             )
-            self.environment_data_polling_manager_thread.start()
+            self.session.proxies.update(proxies or {})
+            retries = retries or Retry(total=3, backoff_factor=0.1)
+
+            self.api_url = api_url if api_url.endswith("/") else f"{api_url}/"
+            self.request_timeout_seconds = request_timeout_seconds
+            self.session.mount(self.api_url, HTTPAdapter(max_retries=retries))
+
+            self.environment_flags_url = f"{self.api_url}flags/"
+            self.identities_url = f"{self.api_url}identities/"
+            self.environment_url = f"{self.api_url}environment-document/"
+
+            self._environment = None
+            if enable_local_evaluation:
+                if not environment_key.startswith("ser."):
+                    raise ValueError(
+                        "In order to use local evaluation, please generate a server key "
+                        "in the environment settings page."
+                    )
+
+                self.environment_data_polling_manager_thread = (
+                    EnvironmentDataPollingManager(
+                        main=self,
+                        refresh_interval_seconds=environment_refresh_interval_seconds,
+                        daemon=True,  # noqa
+                    )
+                )
+                self.environment_data_polling_manager_thread.start()
 
         self._analytics_processor = (
             AnalyticsProcessor(
@@ -164,7 +175,10 @@ def get_identity_segments(
         return [Segment(id=sm.id, name=sm.name) for sm in segment_models]
 
     def update_environment(self):
-        self._environment = self._get_environment_from_api()
+        if self.offline_mode:
+            self._environment = self.offline_handler.get_environment()
+        else:
+            self._environment = self._get_environment_from_api()
 
     def _get_environment_from_api(self) -> EnvironmentModel:
         environment_data = self._get_json_response(self.environment_url, method="GET")

flagsmith/offline_handlers.py

@@ -0,0 +1,22 @@
+import json
+from abc import ABC, abstractmethod
+
+from flag_engine.environments.builders import build_environment_model
+from flag_engine.environments.models import EnvironmentModel
+
+
+class BaseOfflineModeHandler(ABC):
+    @abstractmethod
+    def get_environment(self) -> EnvironmentModel:
+        raise NotImplementedError()
+
+
+class LocalFileStorageHandler(BaseOfflineModeHandler):
+    def __init__(self, environment_document_path: str):
+        with open(environment_document_path) as environment_document:
+            self.environment = build_environment_model(
+                json.loads(environment_document.read())
+            )
+
+    def get_environment(self) -> EnvironmentModel:
+        return self.environment

tests/test_flagsmith.py

@@ -1,5 +1,6 @@
 import json
 import uuid
+from unittest.mock import mock_open, patch
 
 import pytest
 import requests
@@ -8,7 +9,8 @@
 
 from flagsmith import Flagsmith
 from flagsmith.exceptions import FlagsmithAPIError
-from flagsmith.models import DefaultFlag
+from flagsmith.models import DefaultFlag, Flags
+from flagsmith.offline_handlers import LocalFileStorageHandler
 
 
 def test_flagsmith_starts_polling_manager_on_init_if_enabled(mocker, server_api_key):
@@ -378,3 +380,25 @@ def test_initialise_flagsmith_with_proxies():
 
     # Then
     assert flagsmith.session.proxies == proxies
+
+
+def test_offline_mode(environment_json: str) -> None:
+    # Given
+    environment_document_file_path = "some/file/path/environment.json"
+
+    # When
+    with patch("builtins.open", mock_open(read_data=environment_json)) as mock_file:
+        flagsmith = Flagsmith(
+            offline_mode=True,
+            offline_handler=LocalFileStorageHandler(environment_document_file_path),
+        )
+
+        # Then
+        mock_file.assert_called_once_with(environment_document_file_path)
+
+        # and we can request the flags from the client successfully
+        environment_flags: Flags = flagsmith.get_environment_flags()
+        assert environment_flags.is_feature_enabled("some_feature") is True
+
+        identity_flags: Flags = flagsmith.get_identity_flags("identity")
+        assert identity_flags.is_feature_enabled("some_feature") is True