From fb9a8bf04219268eedd261471fa8b89c1583f9e3 Mon Sep 17 00:00:00 2001 From: Lee Chapman Date: Mon, 20 Oct 2025 22:18:54 -0700 Subject: [PATCH 1/2] feat: add exponential backoff for rate limit handling --- src/scraper.py | 102 +++++++++++++++++++++---------------------------- 1 file changed, 44 insertions(+), 58 deletions(-) diff --git a/src/scraper.py b/src/scraper.py index f36ba55..016db21 100644 --- a/src/scraper.py +++ b/src/scraper.py @@ -58,73 +58,65 @@ def _rate_limit(self): time.sleep(delay) self.last_request_time = time.time() + def _fetch_with_backoff(self, url: str, max_retries: int = 5) -> Optional[Dict]: + """Fetch URL with exponential backoff on 429 errors + + Returns either a dict or list depending on the endpoint. + """ + for attempt in range(max_retries): + try: + self._rate_limit() + response = self.session.get(url, timeout=self.timeout) + response.raise_for_status() + return response.json() + except requests.HTTPError as e: + if e.response.status_code == 429: + if attempt < max_retries - 1: + # Exponential backoff: 2^attempt * 2 seconds (2s, 4s, 8s, 16s, 32s) + backoff_delay = (2 ** attempt) * 2 + logger.warning( + f"Rate limited (429). Attempt {attempt + 1}/{max_retries}. " + f"Backing off for {backoff_delay}s before retry..." + ) + time.sleep(backoff_delay) + else: + logger.error(f"Rate limited after {max_retries} attempts: {e}") + return None + else: + logger.error(f"HTTP error {e.response.status_code}: {e}") + return None + except requests.RequestException as e: + logger.error(f"Request failed: {e}") + return None + return None + def get_war_status(self) -> Optional[Dict]: """Fetch current war status""" - try: - self._rate_limit() - response = self.session.get(f"{self.base_url}/war", timeout=self.timeout) - response.raise_for_status() - return response.json() - except requests.RequestException as e: - logger.error(f"Failed to fetch war status: {e}") - return None + return self._fetch_with_backoff(f"{self.base_url}/war") def get_campaign_info(self) -> Optional[List[Dict]]: """Fetch active campaigns information""" - try: - self._rate_limit() - response = self.session.get(f"{self.base_url}/campaigns", timeout=self.timeout) - response.raise_for_status() - return response.json() - except requests.RequestException as e: - logger.error(f"Failed to fetch campaign info: {e}") - return None + result = self._fetch_with_backoff(f"{self.base_url}/campaigns") + return result if isinstance(result, list) else None def get_assignments(self) -> Optional[List[Dict]]: """Fetch current assignments (Major Orders)""" - try: - self._rate_limit() - response = self.session.get(f"{self.base_url}/assignments", timeout=self.timeout) - response.raise_for_status() - return response.json() - except requests.RequestException as e: - logger.error(f"Failed to fetch assignments: {e}") - return None + result = self._fetch_with_backoff(f"{self.base_url}/assignments") + return result if isinstance(result, list) else None def get_dispatches(self) -> Optional[List[Dict]]: """Fetch news dispatches and announcements""" - try: - self._rate_limit() - response = self.session.get(f"{self.base_url}/dispatches", timeout=self.timeout) - response.raise_for_status() - return response.json() - except requests.RequestException as e: - logger.error(f"Failed to fetch dispatches: {e}") - return None + result = self._fetch_with_backoff(f"{self.base_url}/dispatches") + return result if isinstance(result, list) else None def get_planets(self) -> Optional[List[Dict]]: """Fetch all planets information""" - try: - self._rate_limit() - response = self.session.get(f"{self.base_url}/planets", timeout=self.timeout) - response.raise_for_status() - return response.json() - except requests.RequestException as e: - logger.error(f"Failed to fetch planets: {e}") - return None + result = self._fetch_with_backoff(f"{self.base_url}/planets") + return result if isinstance(result, list) else None def get_planet_status(self, planet_index: int) -> Optional[Dict]: """Fetch status of a specific planet""" - try: - self._rate_limit() - response = self.session.get( - f"{self.base_url}/planets/{planet_index}", timeout=self.timeout - ) - response.raise_for_status() - return response.json() - except requests.RequestException as e: - logger.error(f"Failed to fetch planet status for {planet_index}: {e}") - return None + return self._fetch_with_backoff(f"{self.base_url}/planets/{planet_index}") def get_statistics(self) -> Optional[Dict]: """Fetch global game statistics (part of war status)""" @@ -140,14 +132,8 @@ def get_statistics(self) -> Optional[Dict]: def get_planet_events(self) -> Optional[List[Dict]]: """Fetch planet events""" - try: - self._rate_limit() - response = self.session.get(f"{self.base_url}/planet-events", timeout=self.timeout) - response.raise_for_status() - return response.json() - except requests.RequestException as e: - logger.error(f"Failed to fetch planet events: {e}") - return None + result = self._fetch_with_backoff(f"{self.base_url}/planet-events") + return result if isinstance(result, list) else None def get_factions(self) -> Optional[List[Dict]]: """Fetch all factions (from war data)""" From bfee9c0d1bbc182da7bec87b8bd081a47962d87b Mon Sep 17 00:00:00 2001 From: Lee Chapman Date: Mon, 20 Oct 2025 22:23:30 -0700 Subject: [PATCH 2/2] refactor: address rate limiting review feedback --- src/scraper.py | 69 ++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 56 insertions(+), 13 deletions(-) diff --git a/src/scraper.py b/src/scraper.py index 016db21..c7a948f 100644 --- a/src/scraper.py +++ b/src/scraper.py @@ -2,7 +2,7 @@ import logging import time import threading -from typing import Dict, List, Optional +from typing import Dict, List, Optional, Union from src.config import Config logger = logging.getLogger(__name__) @@ -58,22 +58,28 @@ def _rate_limit(self): time.sleep(delay) self.last_request_time = time.time() - def _fetch_with_backoff(self, url: str, max_retries: int = 5) -> Optional[Dict]: + def _fetch_with_backoff(self, url: str, max_retries: int = 5) -> Optional[Union[Dict, List[Dict]]]: """Fetch URL with exponential backoff on 429 errors + Rate limiting is applied once before the first request. + Exponential backoff handles retry delays independently. + Returns either a dict or list depending on the endpoint. """ + # Apply rate limiting once before first request attempt + self._rate_limit() + for attempt in range(max_retries): try: - self._rate_limit() response = self.session.get(url, timeout=self.timeout) response.raise_for_status() return response.json() except requests.HTTPError as e: if e.response.status_code == 429: if attempt < max_retries - 1: - # Exponential backoff: 2^attempt * 2 seconds (2s, 4s, 8s, 16s, 32s) - backoff_delay = (2 ** attempt) * 2 + # Exponential backoff: 5s, 10s, 20s, 40s, 80s + # Aligns with API's 10-second rate limit window + backoff_delay = (2 ** attempt) * 5 logger.warning( f"Rate limited (429). Attempt {attempt + 1}/{max_retries}. " f"Backing off for {backoff_delay}s before retry..." @@ -92,38 +98,70 @@ def _fetch_with_backoff(self, url: str, max_retries: int = 5) -> Optional[Dict]: def get_war_status(self) -> Optional[Dict]: """Fetch current war status""" - return self._fetch_with_backoff(f"{self.base_url}/war") + result = self._fetch_with_backoff(f"{self.base_url}/war") + if result is None: + return None + if isinstance(result, dict): + return result + logger.warning(f"Expected dict from war status endpoint, got {type(result).__name__}") + return None def get_campaign_info(self) -> Optional[List[Dict]]: """Fetch active campaigns information""" result = self._fetch_with_backoff(f"{self.base_url}/campaigns") - return result if isinstance(result, list) else None + if result is None: + return None + if not isinstance(result, list): + logger.warning(f"Expected list from campaigns endpoint, got {type(result).__name__}") + return None + return result def get_assignments(self) -> Optional[List[Dict]]: """Fetch current assignments (Major Orders)""" result = self._fetch_with_backoff(f"{self.base_url}/assignments") - return result if isinstance(result, list) else None + if result is None: + return None + if not isinstance(result, list): + logger.warning(f"Expected list from assignments endpoint, got {type(result).__name__}") + return None + return result def get_dispatches(self) -> Optional[List[Dict]]: """Fetch news dispatches and announcements""" result = self._fetch_with_backoff(f"{self.base_url}/dispatches") - return result if isinstance(result, list) else None + if result is None: + return None + if not isinstance(result, list): + logger.warning(f"Expected list from dispatches endpoint, got {type(result).__name__}") + return None + return result def get_planets(self) -> Optional[List[Dict]]: """Fetch all planets information""" result = self._fetch_with_backoff(f"{self.base_url}/planets") - return result if isinstance(result, list) else None + if result is None: + return None + if not isinstance(result, list): + logger.warning(f"Expected list from planets endpoint, got {type(result).__name__}") + return None + return result def get_planet_status(self, planet_index: int) -> Optional[Dict]: """Fetch status of a specific planet""" - return self._fetch_with_backoff(f"{self.base_url}/planets/{planet_index}") + result = self._fetch_with_backoff(f"{self.base_url}/planets/{planet_index}") + if result is None: + return None + if isinstance(result, dict): + return result + logger.warning(f"Expected dict from planet status endpoint, got {type(result).__name__}") + return None def get_statistics(self) -> Optional[Dict]: """Fetch global game statistics (part of war status)""" try: # Statistics are included in war status, return the statistics subset war_data = self.get_war_status() - if war_data and "statistics" in war_data: + if war_data and isinstance(war_data, dict) and "statistics" in war_data: return war_data["statistics"] return None except requests.RequestException as e: @@ -133,7 +171,12 @@ def get_statistics(self) -> Optional[Dict]: def get_planet_events(self) -> Optional[List[Dict]]: """Fetch planet events""" result = self._fetch_with_backoff(f"{self.base_url}/planet-events") - return result if isinstance(result, list) else None + if result is None: + return None + if not isinstance(result, list): + logger.warning(f"Expected list from planet-events endpoint, got {type(result).__name__}") + return None + return result def get_factions(self) -> Optional[List[Dict]]: """Fetch all factions (from war data)"""