#828 Persistent change id between sessions

Author: bogadisa

src/backend_data_retrieval/data_retrieval_app/external_data_retrieval/cache.py

@@ -0,0 +1,15 @@
+from contextlib import contextmanager
+
+import redis
+
+from data_retrieval_app.external_data_retrieval.config import settings
+
+
+@contextmanager
+def get_cache():
+    cache = redis.from_url(str(settings.CACHE_URI), decode_responses=True)
+
+    try:
+        yield cache
+    finally:
+        cache.close()

src/backend_data_retrieval/data_retrieval_app/external_data_retrieval/config.py

@@ -1,7 +1,9 @@
 from pydantic import (
+    AnyUrl,
     HttpUrl,
     computed_field,
 )
+from pydantic_core import MultiHostUrl
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -20,6 +22,22 @@ def BACKEND_BASE_URL(self) -> HttpUrl:
         else:
             return HttpUrl("http://src-backend-1:8000/api/api_v1")
 
+    REDIS_PORT: int = 6379
+    REDIS_SERVER: str
+    REDIS_CACHE: str = str(0)
+    REDIS_PASSWORD: str = ""
+
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def CACHE_URI(self) -> AnyUrl:
+        return MultiHostUrl.build(
+            scheme="redis",
+            password=self.REDIS_PASSWORD,
+            host=self.REDIS_SERVER,
+            port=self.REDIS_PORT,
+            path=self.REDIS_CACHE,
+        )
+
     MANUAL_NEXT_CHANGE_ID: bool
     NEXT_CHANGE_ID: str
     OATH_ACC_TOKEN_CONTACT_EMAIL: str

src/backend_data_retrieval/data_retrieval_app/external_data_retrieval/data_retrieval/poe_api_handler.py

@@ -9,6 +9,7 @@
 
 import httpx
 import pandas as pd
+import redis
 
 from data_retrieval_app.external_data_retrieval.config import settings
 from data_retrieval_app.external_data_retrieval.data_retrieval.utils import (
@@ -102,7 +103,7 @@ def _json_to_df(self, stashes: list) -> pd.DataFrame | None:
 
         return df
 
-    def _check_stashes(self, stashes: list) -> pd.DataFrame:
+    def _detector_filter(self, stashes: list) -> pd.DataFrame:
         """
         Parameters:
             :param stashes: (list) A list of stash objects
@@ -176,11 +177,7 @@ def _get_latest_change_id(self) -> str:
 
         return next_change_id
 
-    def _follow_stream(
-        self,
-        client: httpx.Client,
-        stop_event: threading.Event,
-    ):
+    def _follow_stream(self, client: httpx.Client, stop_event: threading.Event):
         local_pending = []
         change_id = self.initial_change_id
         weird_errors = 0
@@ -246,7 +243,7 @@ def _follow_stream(
             self.pending_queue.put(None)
 
     @sync_timing_tracker
-    def _process_stream(self) -> pd.DataFrame | None:
+    def _process_stream(self, cache: redis.Redis) -> pd.DataFrame | None:
         i = 0
         stashes = []
         while i < self.mini_batch_size:
@@ -259,26 +256,25 @@ def _process_stream(self) -> pd.DataFrame | None:
             if pending is None:
                 return stashes
 
-            # try:
             obj = json.loads(pending.response.decode("utf-8"))
             stashes.extend(obj["stashes"])
-            # finally:
-            #     pending.response.close()
+
+            cache.set("next_change_id", pending.next_change_id)
 
             i += 1
 
         logger.info("Stashes are ready for processing")
-        wanted_df = self._check_stashes(stashes)
+        wanted_df = self._detector_filter(stashes)
         logger.info("Finished processing the data, waiting for more")
         if wanted_df.empty:
             return None
         return wanted_df
 
-    def _gather_n_checkpoints(self, n: int) -> pd.DataFrame | None:
+    def _gather_n_checkpoints(self, cache: redis.Redis, n: int) -> pd.DataFrame | None:
         df = None
         for _ in range(n):
             start_time = time.perf_counter()
-            wanted_df = self._process_stream()
+            wanted_df = self._process_stream(cache)
             end_time = time.perf_counter()
 
             time_per_mini_batch = end_time - start_time
@@ -299,9 +295,16 @@ def _gather_n_checkpoints(self, n: int) -> pd.DataFrame | None:
         return df
 
     def initialize_data_stream_threads(
-        self, executor: ThreadPoolExecutor, stop_event: threading.Event
+        self,
+        executor: ThreadPoolExecutor,
+        stop_event: threading.Event,
+        cache: redis.Redis,
     ):
-        self.initial_change_id = self._get_latest_change_id()
+        self.initial_change_id = cache.get("next_change_id")
+        if self.initial_change_id is None or settings.MANUAL_NEXT_CHANGE_ID:
+            logger.info("Using manually set change id")
+            self.initial_change_id = self._get_latest_change_id()
+
         self.rate_limiter = RateLimiter()
         self.mini_batch_size = settings.MINI_BATCH_SIZE
         self.pending_queue = Queue(maxsize=self.mini_batch_size)
@@ -318,11 +321,12 @@ def initialize_data_stream_threads(
         logging.getLogger("httpx").setLevel(logging.WARNING)
         return executor.submit(self._follow_stream, client, stop_event)
 
-    def dump_stream(self) -> Iterator[pd.DataFrame]:
+    def dump_stream(self, cache: redis.Redis) -> Iterator[pd.DataFrame]:
         time.sleep(5)  # Waits for the listening threads to have time to start up.
         while True:
             logger.info("Waiting for data from the stream")
             df = self._gather_n_checkpoints(
+                cache,
                 n=settings.N_CHECKPOINTS_PER_TRANSFORMATION,
             )
             if df is None:

src/backend_data_retrieval/data_retrieval_app/external_data_retrieval/main.py

@@ -10,7 +10,9 @@
 from typing import Any
 
 import pandas as pd
+import redis
 
+from data_retrieval_app.external_data_retrieval.cache import get_cache
 from data_retrieval_app.external_data_retrieval.config import settings
 from data_retrieval_app.external_data_retrieval.data_retrieval.currency_api_handler import (
     CurrencyAPIHandler,
@@ -208,7 +210,7 @@ def _initialize_data_stream_threads(
             executor, listeners, has_crashed
         )
 
-    def _follow_data_dump_stream(self):
+    def _follow_data_dump_stream(self, cache: redis.Redis):
         current_hours = find_hours_since_launch(self.leagues)
         # Only need to refer to one league to see when a new hour starts
         current_hour = current_hours[self.leagues[0]["leagueId"]]
@@ -217,7 +219,7 @@ def _follow_data_dump_stream(self):
         modifier_dfs = self._get_modifiers()
         item_base_types = self._get_item_base_types()
         currency_df = self._get_new_currency_data(current_hours)
-        get_df = self.poe_api_handler.dump_stream()
+        get_df = self.poe_api_handler.dump_stream(cache)
         while current_hour < next_hour:
             df = next(get_df)
             split_dfs = self._categorize_new_items(df)
@@ -239,16 +241,15 @@ def retrieve_data(self):
         logger.info("Initiating data stream.")
         max_workers = 2
         stop_event = threading.Event()
-        with ThreadPoolExecutor(max_workers=max_workers) as executor:
-            # futures = self._initialize_data_stream_threads(
-            #     executor, listeners=listeners
-            # )
+        with ThreadPoolExecutor(
+            max_workers=max_workers
+        ) as executor, get_cache() as cache:
             futures = {}
             listener_future = self.poe_api_handler.initialize_data_stream_threads(
-                executor, stop_event
+                executor, stop_event, cache
             )
             futures[listener_future] = "listener"
-            follow_future = executor.submit(self._follow_data_dump_stream)
+            follow_future = executor.submit(self._follow_data_dump_stream, cache)
             futures[follow_future] = "data_processing"
             logger.info("Waiting for futures to crash.")
             finished = False

src/docker-compose.yml

@@ -30,7 +30,7 @@ services:
       retries: 5
 
   cache:
-    image: redis:7.4.0
+    image: redis:7.4
     restart: always
     networks:
       - ${TRAEFIK_PUBLIC_NETWORK?Variable not set}
@@ -49,7 +49,7 @@ services:
       interval: 1s
       timeout: 3s
       retries: 5
-    command: redis-server --save 20 1 --loglevel warning --maxmemory 1000mb --requirepass ${REDIS_PASSWORD?Variable not set}
+    command: redis-server --save 20 1 --loglevel warning --maxmemory 1000mb --requirepass ${REDIS_PASSWORD?Variable not set} --appendonly yes
 
   backend:
     image: "${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG-latest}"