refactor method to use a queue

Author: Herraj

merino/providers/suggest/finance/backends/polygon/backend.py

@@ -94,6 +94,8 @@ class PolygonBackend(FinanceBackend):
     url_single_ticker_snapshot: str
     url_single_ticker_overview: str
     filemanager: PolygonFilemanager
+    cache_refresh_task: asyncio.Task
+    cache_refresh_task_queue: asyncio.Queue
 
     def __init__(
         self,
@@ -133,6 +135,9 @@ def __init__(
             SCRIPT_ID_BULK_WRITE_TICKERS, LUA_SCRIPT_CACHE_BULK_WRITE_TICKERS
         )
 
+        self.cache_refresh_task_queue = asyncio.Queue()
+        self.cache_refresh_task = asyncio.create_task(self.refresh_ticker_cache_entries())
+
     async def get_snapshots(self, tickers: list[str]) -> list[TickerSnapshot]:
         """Get snapshots for the list of tickers."""
         snapshots: list[TickerSnapshot] = []
@@ -368,26 +373,26 @@ async def store_snapshots_in_cache(self, snapshots: list[TickerSnapshot]) -> Non
         )
 
     async def refresh_ticker_cache_entries(
-        self, tickers: list[str], *, await_store: bool = False
+        self
     ) -> None:
         """Refresh ticker snapshot in cache. Fetches new snapshot from upstream API and
         fires a background task to write it to the cache.
-
-        Note: Only awaits the cache write process if await_store is true. Used only for testing.
         """
-        snapshots = await self.get_snapshots(tickers)
+        while True:
+            # Get the tickers list from the queue
+            # NOTE: get_nowait() can throw QueueEmpty exception
+            tickers: list[str] = self.cache_refresh_task_queue.get_nowait()
+            snapshots = await self.get_snapshots(tickers)
 
-        # early exit if no snapshots returned.
-        if len(snapshots) == 0:
-            return
+            # Early exit if no snapshots returned.
+            # Although, the store method also has the same check.
+            if len(snapshots) == 0:
+                return
 
-        # this parameter is only used for testing.
-        if await_store:
             await self.store_snapshots_in_cache(snapshots)
-        else:
-            task = asyncio.create_task(self.store_snapshots_in_cache(snapshots))
-            # consume/log
-            task.add_done_callback(lambda t: t.exception())
+
+            # notify queue that the task is done
+            self.cache_refresh_task_queue.task_done()
 
     # TODO @herraj add unit tests for this
     def _parse_cached_data(