opt: improvements on background app refresh and tmdb data fetch (#7)

Author: TimilsinaBimal

app/core/app.py

@@ -37,7 +37,7 @@ async def lifespan(app: FastAPI):
     global catalog_updater
 
     # Startup
-    if settings.AUTO_UPDATE_CATALOGS and settings.CATALOG_REFRESH_INTERVAL_SECONDS > 0:
+    if settings.AUTO_UPDATE_CATALOGS and settings.CATALOG_REFRESH_INTERVAL_SECONDS > 3600:  # minimum 1 hour
         catalog_updater = BackgroundCatalogUpdater(interval_seconds=settings.CATALOG_REFRESH_INTERVAL_SECONDS)
         catalog_updater.start()
         logger.info(

app/services/catalog_updater.py

@@ -1,6 +1,8 @@
 import asyncio
 from typing import Any
 
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from apscheduler.triggers.interval import IntervalTrigger
 from loguru import logger
 
 from app.services.catalog import DynamicCatalogService
@@ -38,21 +40,28 @@ class BackgroundCatalogUpdater:
 
     def __init__(self, interval_seconds: int) -> None:
         self.interval_seconds = max(60, interval_seconds)
-        self._task: asyncio.Task | None = None
-        self._stop_event = asyncio.Event()
+        self.scheduler = AsyncIOScheduler()
 
     def start(self) -> None:
-        if self._task is not None:
+        if self.scheduler.running:
             return
-        self._stop_event.clear()
-        self._task = asyncio.create_task(self._run())
+
+        logger.info(f"Starting background catalog updater. Interval: {self.interval_seconds}s")
+        self.scheduler.add_job(
+            self.refresh_all_tokens,
+            trigger=IntervalTrigger(seconds=self.interval_seconds),
+            id="catalog_refresh",
+            replace_existing=True,
+            max_instances=1,  # Prevent new job from starting if previous one is still running
+            coalesce=True,  # If multiple runs are missed, only run once
+        )
+        self.scheduler.start()
 
     async def stop(self) -> None:
-        if self._task is None:
-            return
-        self._stop_event.set()
-        await self._task
-        self._task = None
+        if self.scheduler.running:
+            logger.info("Stopping background catalog updater...")
+            self.scheduler.shutdown(wait=True)  # Wait for running jobs to complete
+            logger.info("Background catalog updater stopped.")
 
     async def refresh_all_tokens(self) -> None:
         """Refresh catalogs for all tokens concurrently with a semaphore."""
@@ -89,18 +98,6 @@ async def _update_safe(key: str, payload: dict[str, Any]) -> None:
         except Exception as exc:
             logger.error(f"Catalog refresh scan failed: {exc}", exc_info=True)
 
-    async def _run(self) -> None:
-        logger.info(f"Background catalog updater started. Interval: {self.interval_seconds}s")
-        try:
-            while not self._stop_event.is_set():
-                await self.refresh_all_tokens()
-                try:
-                    await asyncio.wait_for(self._stop_event.wait(), timeout=self.interval_seconds)
-                except TimeoutError:
-                    continue
-        finally:
-            logger.info("Background catalog updater stopped.")
-
     @staticmethod
     def _has_credentials(payload: dict[str, Any]) -> bool:
         return bool(payload.get("authKey") or (payload.get("username") and payload.get("password")))

pyproject.toml

@@ -5,6 +5,7 @@ description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.10"
 dependencies = [
+    "apscheduler>=3.11.1",
     "async-lru>=2.0.5",
     "cachetools>=6.2.2",
     "cryptography>=46.0.3",

requirements.txt

@@ -7,4 +7,5 @@ loguru>=0.7.2
 cachetools>=6.2.2
 redis>=5.0.1
 cryptography>=41.0.0
-async-lru>=2.0.4
+async-lru>=2.0.5
+apscheduler>=3.11.1