Added load_snapshot in-memory from remote API

.github/workflows/master.yml

@@ -31,7 +31,7 @@ jobs:
         run: pipenv run pytest -v --cov=./switcher_client --cov-report xml
 
       - name: SonarCloud Scan
-        uses: sonarsource/sonarqube-scan-action@v5.2.0
+        uses: sonarsource/sonarqube-scan-action@v5.3.0
         env:
           SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
         if: env.SONAR_TOKEN != ''

.github/workflows/sonar.yml

@@ -47,7 +47,7 @@ jobs:
         run: pipenv run pytest -v --cov=./switcher_client --cov-report xml
 
       - name: SonarCloud Scan
-        uses: sonarsource/sonarqube-scan-action@v5.2.0
+        uses: sonarsource/sonarqube-scan-action@v5.3.0
         env:
           SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
         if: env.SONAR_TOKEN != ''

Pipfile

@@ -4,12 +4,12 @@ verify_ssl = true
 name = "pypi"
 
 [packages]
-requests = "==2.32.4"
+requests = "==2.32.5"
 
 # Managed dependencies and security patches
 urllib3 = ">=2.5.0"
 
 [dev-packages]
 pytest = "==8.4.1"
 pytest-cov = "==6.2.1"
-responses = "==0.25.7"
+responses = "==0.25.8"

switcher_client/client.py

@@ -1,10 +1,10 @@
 from typing import Optional
 
-from switcher_client.lib.globals.global_snapshot import GlobalSnapshot
+from switcher_client.lib.globals.global_snapshot import GlobalSnapshot, LoadSnapshotOptions
 from switcher_client.lib.remote_auth import RemoteAuth
 from switcher_client.lib.globals.global_context import Context, ContextOptions
 from switcher_client.lib.globals.global_context import DEFAULT_ENVIRONMENT
-from switcher_client.lib.snapshot_loader import load_domain
+from switcher_client.lib.snapshot_loader import load_domain, validate_snapshot
 from switcher_client.lib.utils import get
 from switcher_client.switcher import Switcher
 
@@ -62,15 +62,39 @@ def get_switcher(key: Optional[str] = None) -> Switcher:
 
 
     @staticmethod
-    def load_snapshot() -> int:
+    def load_snapshot(options: Optional[LoadSnapshotOptions] = None) -> int:
         """ Load Domain from snapshot """
-
+        snapshot_options = get(options, LoadSnapshotOptions())
+
         GlobalSnapshot.init(load_domain(
             get(Client.context.options.snapshot_location, ''),
             get(Client.context.environment, DEFAULT_ENVIRONMENT)
         ))
 
+        if Client._is_check_snapshot_available(snapshot_options.fetch_remote):
+            Client.check_snapshot()
+
         return Client.snapshot_version()
+
+    @staticmethod
+    def check_snapshot():
+        """ Verifies if the current snapshot file is updated
+            Return true if an update has been made
+        """
+
+        if RemoteAuth.is_token_expired():
+            RemoteAuth.auth()
+
+        snapshot = validate_snapshot(
+            context=Client.context,
+            snapshot_version=Client.snapshot_version(),
+        )
+
+        if snapshot is not None:
+            GlobalSnapshot.init(snapshot)
+            return True
+
+        return False
 
     @staticmethod
     def snapshot_version() -> int:
@@ -80,4 +104,8 @@ def snapshot_version() -> int:
         if snapshot is None:
             return 0
 
-        return snapshot.data.domain.version
+        return snapshot.data.domain.version
+
+    @staticmethod
+    def _is_check_snapshot_available(fetch_remote = False) -> bool:
+        return Client.snapshot_version() == 0 and (fetch_remote or not Client.context.options.local)

switcher_client/errors/__init__.py

@@ -1,9 +1,12 @@
-class RemoteAuthError(Exception):
+class RemoteError(Exception):
     def __init__(self, message):
         self.message = message
         super().__init__(self.message)
 
-class RemoteCriteriaError(Exception):
+class RemoteAuthError(RemoteError):
     def __init__(self, message):
-        self.message = message
-        super().__init__(self.message)
+        super().__init__(message)
+
+class RemoteCriteriaError(RemoteError):
+    def __init__(self, message):
+        super().__init__(message)

switcher_client/lib/globals/global_snapshot.py

@@ -12,4 +12,9 @@ def clear():
 
     @staticmethod
     def snapshot() -> Snapshot | None:
-        return GlobalSnapshot.snapshotStore
+        return GlobalSnapshot.snapshotStore
+
+class LoadSnapshotOptions:
+    def __init__(self, fetch_remote: bool = False, watch_snapshot: bool = False):
+        self.fetch_remote = fetch_remote
+        self.watch_snapshot = watch_snapshot

switcher_client/lib/remote.py

@@ -1,18 +1,18 @@
+import json
 import requests
 from typing import Optional
 
-from switcher_client.errors import RemoteAuthError
+from switcher_client.errors import RemoteAuthError, RemoteError
 from switcher_client.errors import RemoteCriteriaError
-from switcher_client.lib.globals.global_context import Context
+from switcher_client.lib.globals.global_context import DEFAULT_ENVIRONMENT, Context
 from switcher_client.lib.types import ResultDetail
+from switcher_client.lib.utils import get
 from switcher_client.switcher_data import SwitcherData
 
 class Remote:
 
     @staticmethod
     def auth(context: Context):
-        """ Authenticate """
-
         url = f'{context.url}/criteria/auth'
         response = Remote.do_post(url, {
             'domain': context.domain,
@@ -31,7 +31,6 @@ def auth(context: Context):
     @staticmethod
     def check_criteria(
         token: Optional[str], context: Context, switcher: SwitcherData) -> ResultDetail:
-        """ Check criteria """
 
         url = f'{context.url}/criteria?showReason={str(switcher._show_details).lower()}&key={switcher._key}'
         entry = Remote.__get_entry(switcher._input)
@@ -46,12 +45,55 @@ def check_criteria(
             )
 
         raise RemoteCriteriaError(f'[check_criteria] failed with status: {response.status_code}')
+
+    @staticmethod
+    def check_snapshot_version(token: Optional[str], context: Context, snapshot_version: int) -> bool:
+        url = f'{context.url}/criteria/snapshot_check/{snapshot_version}'
+        response = Remote.do_get(url, Remote.get_header(token))
+
+        if response.status_code == 200:
+            return response.json().get('status', False)
+
+        raise RemoteError(f'[check_snapshot_version] failed with status: {response.status_code}')
+
+    @staticmethod
+    def resolve_snapshot(token: Optional[str], context: Context) -> str | None:
+        domain = get(context.domain, '')
+        environment = get(context.environment, DEFAULT_ENVIRONMENT)
+        component = get(context.component, '')
+
+        data = {
+            "query": f"""
+                query domain {{
+                    domain(name: "{domain}", environment: "{environment}", _component: "{component}") {{
+                    name version activated
+                    group {{ name activated
+                        config {{ key activated
+                            strategies {{ strategy activated operation values }}
+                            relay {{ type  activated }}
+                            components
+                        }}
+                    }}
+                }}
+            }}
+            """
+        }
+
+        response = Remote.do_post(f'{context.url}/graphql', data, Remote.get_header(token))
 
+        if response.status_code == 200:
+            return json.dumps(response.json(), indent=4)
+
+        raise RemoteError(f'[resolve_snapshot] failed with status: {response.status_code}')
+
     @staticmethod
     def do_post(url, data, headers) -> requests.Response:
-        """ Perform a POST request """
         return requests.post(url, json=data, headers=headers)
-
+
+    @staticmethod
+    def do_get(url, headers=None) -> requests.Response:
+        return requests.get(url, headers=headers)
+
     @staticmethod
     def get_header(token: Optional[str]):
         return {

switcher_client/lib/snapshot_loader.py

@@ -1,7 +1,10 @@
 import json
 import os
 
-from switcher_client.lib.types import Snapshot, SnapshotData, Domain, Group, Config, StrategyConfig, Relay
+from switcher_client.lib.globals.global_auth import GlobalAuth
+from switcher_client.lib.globals.global_context import Context
+from switcher_client.lib.remote import Remote
+from switcher_client.lib.types import Snapshot
 
 def load_domain(snapshot_location: str, environment: str):
     """ Load Domain from snapshot file """
@@ -27,74 +30,24 @@ def load_domain(snapshot_location: str, environment: str):
         with open(snapshot_file, 'r') as file:
             json_data = json.load(file)
 
-    snapshot = Snapshot()
-    snapshot.data = SnapshotData()
-    snapshot.data.domain = _parse_domain(json_data['data']['domain'])
-
-    return snapshot
-
-def _parse_domain(domain_data: dict) -> Domain:
-    """ Parse domain data from JSON """
+    snapshot = Snapshot(json_data.get('data', {}))
 
-    domain = Domain()
-    domain.name = domain_data.get('name')
-    domain.activated = domain_data.get('activated')
-    domain.version = domain_data.get('version', 0)
-
-    if 'group' in domain_data and domain_data['group']:
-        domain.group = []
-        for group_data in domain_data['group']:
-            domain.group.append(_parse_group(group_data))
-
-    return domain
-
-def _parse_group(group_data: dict) -> Group:
-    """ Parse group data from JSON """
-
-    group = Group()
-    group.name = group_data.get('name')
-    group.activated = group_data.get('activated')
-
-    if 'config' in group_data and group_data['config']:
-        group.config = []
-        for config_data in group_data['config']:
-            group.config.append(_parse_config(config_data))
-
-    return group
-
-def _parse_config(config_data: dict) -> Config:
-    """ Parse config data from JSON """
-
-    config = Config()
-    config.key = config_data.get('key')
-    config.activated = config_data.get('activated')
-
-    if 'strategies' in config_data and config_data['strategies']:
-        config.strategies = []
-        for strategy_data in config_data['strategies']:
-            config.strategies.append(_parse_strategy(strategy_data))
-
-    if 'relay' in config_data and config_data['relay']:
-        config.relay = _parse_relay(config_data['relay'])
-
-    return config
+    return snapshot
 
-def _parse_strategy(strategy_data: dict) -> StrategyConfig:
-    """ Parse strategy data from JSON """
+def validate_snapshot(
+    context: Context,
+    snapshot_version: int
+) -> Snapshot | None:
+    """ Validate the snapshot data """
 
-    strategy = StrategyConfig()
-    strategy.strategy = strategy_data.get('strategy')
-    strategy.activated = strategy_data.get('activated')
-    strategy.operation = strategy_data.get('operation')
-    strategy.values = strategy_data.get('values')
+    status = Remote.check_snapshot_version(
+        token=GlobalAuth.get_token(), 
+        context=context,
+        snapshot_version=snapshot_version)
 
-    return strategy
-
-def _parse_relay(relay_data: dict) -> Relay:
-    """ Parse relay data from JSON """
-
-    relay = Relay()
-    relay.type = relay_data.get('type')
-    relay.activated = relay_data.get('activated')
+    if not status:
+        snapshot_str = Remote.resolve_snapshot(GlobalAuth.get_token(), context)
+        graphql_response = json.loads(snapshot_str or '{}')
+        return Snapshot(graphql_response.get('data', '{}'))
 
-    return relay
+    return None