chore: refactored SnapshotAutoUpdater to use instance-based approach

switcher_client/client.py

@@ -22,6 +22,7 @@ class SwitcherOptions:
 class Client:
     _context: Context = Context.empty()
     _switcher: dict[str, Switcher] = {}
+    _snapshot_auto_updater: SnapshotAutoUpdater = SnapshotAutoUpdater()
     _snapshot_watcher: SnapshotWatcher = SnapshotWatcher()
 
     @staticmethod
@@ -149,7 +150,7 @@ def schedule_snapshot_auto_update(interval: Optional[int] = None,
 
         if Client._context.options.snapshot_auto_update_interval is not None and \
             Client._context.options.snapshot_auto_update_interval > 0:
-            SnapshotAutoUpdater.schedule(
+            Client._snapshot_auto_updater.schedule(
                 interval=Client._context.options.snapshot_auto_update_interval,
                 check_snapshot=Client.check_snapshot,
                 callback=callback
@@ -158,7 +159,7 @@ def schedule_snapshot_auto_update(interval: Optional[int] = None,
     @staticmethod
     def terminate_snapshot_auto_update():
         """ Terminate Snapshot auto update """
-        SnapshotAutoUpdater.terminate()
+        Client._snapshot_auto_updater.terminate()
 
     @staticmethod
     def watch_snapshot(callback: Optional[dict] = None) -> None:

switcher_client/lib/snapshot_auto_updater.py

@@ -4,11 +4,13 @@
 from typing import Callable, Optional
 
 class SnapshotAutoUpdater:
-    _timer_thread: Optional[threading.Thread] = None
-    _stop_event: Optional[threading.Event] = None
+    """ Schedules periodic snapshot updates in a background thread """
 
-    @staticmethod
-    def schedule(interval: int, check_snapshot: Callable[[], bool], callback: Callable[[Optional[Exception], bool], None]) -> None:
+    def __init__(self):
+        self._timer_thread: Optional[threading.Thread] = None
+        self._stop_event: Optional[threading.Event] = None
+
+    def schedule(self, interval: int, check_snapshot: Callable[[], bool], callback: Callable[[Optional[Exception], bool], None]) -> None:
         """
         Schedule periodic snapshot updates in a background thread.
 
@@ -17,34 +19,32 @@ def schedule(interval: int, check_snapshot: Callable[[], bool], callback: Callab
         :param callback: Callback function called with (error, updated) after each check
         """
 
-        SnapshotAutoUpdater.terminate()
-        SnapshotAutoUpdater._stop_event = threading.Event() 
-        
-        SnapshotAutoUpdater._timer_thread = threading.Thread(
-            target=SnapshotAutoUpdater._update_worker,
+        self.terminate()
+        self._stop_event = threading.Event()
+
+        self._timer_thread = threading.Thread(
+            target=self._update_worker,
             args=(interval, check_snapshot, callback),
             daemon=True,
             name="SnapshotAutoUpdater"
         )
-        SnapshotAutoUpdater._timer_thread.start()
+        self._timer_thread.start()
 
-    @staticmethod
-    def terminate() -> None:
+    def terminate(self) -> None:
         """
         Terminate the scheduled snapshot auto-update thread gracefully.
         """
-        if SnapshotAutoUpdater._stop_event is not None:
-            SnapshotAutoUpdater._stop_event.set()
-
-        if SnapshotAutoUpdater._timer_thread is not None and SnapshotAutoUpdater._timer_thread.is_alive():
-            SnapshotAutoUpdater._timer_thread.join(timeout=5.0)
-
-        SnapshotAutoUpdater._timer_thread = None
-        SnapshotAutoUpdater._stop_event = None
+        if self._stop_event is not None:
+            self._stop_event.set()
+
+        if self._timer_thread is not None and self._timer_thread.is_alive():
+            self._timer_thread.join(timeout=5.0)
 
-    @staticmethod
-    def _update_worker(interval: int, check_snapshot: Callable[[], bool], callback: Callable[[Optional[Exception], bool], None]) -> None:
-        stop_event = SnapshotAutoUpdater._stop_event
+        self._timer_thread = None
+        self._stop_event = None
+
+    def _update_worker(self, interval: int, check_snapshot: Callable[[], bool], callback: Callable[[Optional[Exception], bool], None]) -> None:
+        stop_event = self._stop_event
 
         time.sleep(interval) # delay start
         while stop_event is not None and not stop_event.is_set():
@@ -53,5 +53,5 @@ def _update_worker(interval: int, check_snapshot: Callable[[], bool], callback:
                 callback(None, updated)
             except Exception as error:
                 callback(error, False)
-            
+
             stop_event.wait(interval)