Introduce caching for LeetCode API interactions

Added a CachedLeetCodeAPI class to improve performance by caching API responses. Updated PracticeHandler to utilize cached API methods instead of direct API calls. Removed unused fields from LeetCode API queries to streamline data fetching.

Author: nicholasadamou

api/CachedLeetCodeAPI.py

@@ -0,0 +1,110 @@
+import json
+import hashlib
+import tempfile
+from typing import Any, Callable, Optional
+from pathlib import Path
+import time
+
+from api.LeetCodeAPI import LeetCodeAPI
+from utils.logger import log, LogLevel
+
+
+class CachedLeetCodeAPI:
+    def __init__(self, cache_dir: Optional[str] = None, cache_expiry: int = 3600):
+        """
+        Initialize the CachedLeetCodeAPI with caching functionality.
+        :param cache_dir: Directory to store cached API results. Defaults to OS temp directory if None.
+        :param cache_expiry: Expiry time for cache in seconds (default: 1 hour).
+        """
+        # Use the system's temporary directory if no cache directory is provided
+        self.cache_dir = Path(cache_dir or tempfile.gettempdir()) / "cached_leetcode_api"
+        self.cache_expiry = cache_expiry  # Expiry time in seconds
+        self.cache_dir.mkdir(parents=True, exist_ok=True)  # Create cache directory if it doesn't exist
+
+        self.api = LeetCodeAPI()  # Delegate actual API calls to existing LeetCodeAPI class
+
+    def _get_cache_key(self, unique_id: str) -> str:
+        """
+        Generate a cache key based on a unique identifier (e.g., query or API parameters).
+        :param unique_id: Unique identifier for the API request (e.g., slug, query, variable JSON).
+        :return: Cache filename based on the MD5 hash of the unique ID.
+        """
+        return hashlib.md5(unique_id.encode('utf-8')).hexdigest()
+
+    def _read_from_cache(self, cache_key: str) -> Any:
+        """
+        Read data from cache if it exists and is valid.
+        :param cache_key: Key corresponding to the cached data.
+        :return: Cached data, or None if cache is invalid or missing.
+        """
+        cache_file = self.cache_dir / cache_key
+        if cache_file.exists():
+            # Check expiry
+            if time.time() - cache_file.stat().st_mtime <= self.cache_expiry:
+                with open(cache_file, "r") as f:
+                    return json.load(f)
+
+        return None  # Cache miss or expired
+
+    def _write_to_cache(self, cache_key: str, data: Any) -> None:
+        """
+        Write data to cache.
+        :param cache_key: Key corresponding to the cached data.
+        :param data: Data to cache.
+        """
+        cache_file = self.cache_dir / cache_key
+        with open(cache_file, "w") as f:
+            json.dump(data, f)
+
+    def _fetch_with_cache(self, fetch_func: Callable, unique_id: str, *args, **kwargs) -> Any:
+        """
+        Fetch data with caching.
+        :param fetch_func: Function to fetch data if cache is not available.
+        :param unique_id: Unique identifier for the request (e.g., API parameters).
+        :param args: Positional arguments to pass to the fetch function.
+        :param kwargs: Keyword arguments to pass to the fetch function.
+        :return: Fetched or cached data.
+        """
+        cache_key = self._get_cache_key(unique_id)
+
+        # Check cache first
+        cached_data = self._read_from_cache(cache_key)
+        if cached_data is not None:
+            log(f"✅ Cache hit for {unique_id}", LogLevel.DEBUG)
+            return cached_data
+
+        # Cache miss, call the API
+        log(f"❌ Cache miss for {unique_id}. Fetching from API...", LogLevel.DEBUG)
+        api_data = fetch_func(*args, **kwargs)
+
+        # Save API data to cache
+        self._write_to_cache(cache_key, api_data)
+        return api_data
+
+    # Cached versions of API methods
+    def fetch_problems(self, *args, **kwargs):
+        """
+        Cached version of fetch_problems
+        """
+        unique_id = f"fetch_problems-{json.dumps([args, kwargs], sort_keys=True)}"  # Unique key based on args
+        return self._fetch_with_cache(self.api.fetch_problems, unique_id, *args, **kwargs)
+
+    def fetch_daily_challenge(self, *args, **kwargs):
+        """
+        Cached version of fetch_daily_challenge
+        """
+        return self._fetch_with_cache(self.api.fetch_daily_challenge, "fetch_daily_challenge")
+
+    def fetch_problem(self, problem_slug: str, *args, **kwargs):
+        """
+        Cached version of fetch_problem
+        """
+        unique_id = f"fetch_problem-{problem_slug}"
+        return self._fetch_with_cache(self.api.fetch_problem, unique_id, problem_slug, *args, **kwargs)
+
+    def get_study_plan(self, slug: str, *args, **kwargs):
+        """
+        Cached version of get_study_plan
+        """
+        unique_id = f"get_study_plan-{slug}"
+        return self._fetch_with_cache(self.api.get_study_plan, unique_id, slug, *args, **kwargs)

api/LeetCodeAPI.py

@@ -50,7 +50,6 @@ def fetch_problems(
               acRate
               difficulty
               content
-              frontendQuestionId: questionFrontendId
               title
               titleSlug
               codeSnippets {
@@ -127,17 +126,9 @@ def fetch_daily_challenge(self) -> Dict[str, Any]:
             question {
               titleSlug
               title
-              translatedTitle
               content
               acRate
               difficulty
-              freqBar
-              frontendQuestionId: questionFrontendId
-              isFavor
-              paidOnly: isPaidOnly
-              status
-              hasVideoSolution
-              hasSolution
               codeSnippets {
                 lang
                 code
@@ -177,7 +168,6 @@ def fetch_problem(self, problem_slug: str) -> Dict[str, Any]:
         query = """
         query getQuestionDetails($titleSlug: String!) {
             question(titleSlug: $titleSlug) {
-                questionId
                 title
                 titleSlug
                 content
@@ -227,52 +217,14 @@ def get_study_plan(self, slug: str) -> Dict[str, Any]:
           studyPlanV2Detail(planSlug: $slug) {
             slug
             name
-            highlight
-            staticCoverPicture
-            colorPalette
-            threeDimensionUrl
             description
-            premiumOnly
-            needShowTags
-            awardDescription
-            defaultLanguage
-            award {
-              name
-              config {
-                icon
-                iconGif
-                iconGifBackground
-              }
-            }
-            relatedStudyPlans {
-              cover
-              highlight
-              name
-              slug
-              premiumOnly
-            }
             planSubGroups {
               slug
               name
-              premiumOnly
               questionNum
               questions {
-                translatedTitle
-                titleSlug
                 title
-                questionFrontendId
-                paidOnly
-                id
-                difficulty
-                hasOfficialSolution
-                topicTags {
-                  slug
-                  name
-                }
-                solutionInfo {
-                  solutionSlug
-                  solutionTopicId
-                }
+                titleSlug
               }
             }
           }

handlers/CacheHandler.py

@@ -0,0 +1,3 @@
+from api.CachedLeetCodeAPI import CachedLeetCodeAPI
+
+cached_api = CachedLeetCodeAPI(cache_expiry=3600)

handlers/PracticeHandler.py

@@ -5,8 +5,9 @@
 from typing import List, Dict, Any, Optional
 
 from utils.logger import log, LogLevel
+
 from handlers.SolutionHandler import SolutionHandler
-from api.LeetCodeAPI import LeetCodeAPI
+from handlers.CacheHandler import cached_api
 
 difficulty_map = {
     "easy": "Easy",
@@ -17,7 +18,7 @@
 
 class PracticeMode:
     def __init__(self):
-        self.leetcode_api = LeetCodeAPI()
+        pass
 
     def handle(self, args):
         raise NotImplementedError("This method should be implemented by subclasses.")
@@ -93,7 +94,7 @@ def get_random_problem(
         :param difficulties: Difficulty levels of the problems (e.g., "Easy", "Medium", "Hard").
         :return: A random problem dictionary or None if no problems are found.
         """
-        problems = self.leetcode_api.fetch_problems(
+        problems = cached_api.fetch_problems(
             limit=1000, difficulties=difficulties
         )
 
@@ -109,7 +110,7 @@ def get_study_plan_problems(self, slug: str) -> List[Dict[str, Any]]:
         :param slug: The slug of the study plan (e.g., "leetcode-75").
         :return: A list of dictionaries, each containing details of a problem.
         """
-        study_plan = self.leetcode_api.get_study_plan(slug)
+        study_plan = cached_api.get_study_plan(slug)
 
         if not study_plan:
             raise Exception(f"❌ Study plan not found for slug: {slug}")
@@ -137,7 +138,10 @@ def get_random_study_plan_problem(self, slug: str) -> Optional[Dict[str, Any]]:
             return None
 
         random_index = random.randint(0, len(problems) - 1)
-        return problems[random_index]
+        problem = problems[random_index]
+
+        # Get the problem in the right format
+        return cached_api.fetch_problem(problem["titleSlug"])
 
 
 class RandomProblemMode(PracticeMode):
@@ -170,7 +174,7 @@ def handle(self, args):
 
         try:
             # Use the LeetCodeAPI's fetch_daily_challenge method
-            daily_challenge = self.leetcode_api.fetch_daily_challenge()
+            daily_challenge = cached_api.fetch_daily_challenge()
             difficulty_label = difficulty_map[
                 daily_challenge["question"]["difficulty"].lower()
             ]
@@ -196,7 +200,7 @@ def handle(self, args):
 
         for slug in args["problems"]:
             try:
-                problem = self.leetcode_api.fetch_problem(slug.strip())
+                problem = cached_api.fetch_problem(slug.strip())
 
                 if not problem:
                     log(f"Problem with slug '{slug}' not found.", LogLevel.ERROR)
@@ -225,9 +229,6 @@ def handle(self, args):
                 log("No problems found for the selected study plan.", LogLevel.ERROR)
                 return
 
-            # Get the problem in the right format
-            problem = self.leetcode_api.fetch_problem(problem["titleSlug"])
-
             difficulty_label = difficulty_map[problem["difficulty"].lower()]
             url = f"https://leetcode.com/problems/{problem['titleSlug']}"
 

utils/watch_and_submit.py

@@ -2,6 +2,7 @@
 import time
 
 from api.LeetCodeAPI import LeetCodeAPI
+
 from utils.timer import start_timer
 from utils.logger import log, LogLevel