Skip to content

feat: add exponential backoff for rate limit handling #9

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
surrealwolf merged 2 commits into from
Oct 21, 2025
Merged

feat: add exponential backoff for rate limit handling #9

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fb9a8bf
feat: add exponential backoff for rate limit handling
surrealwolf Oct 21, 2025
bfee9c0
refactor: address rate limiting review feedback
surrealwolf Oct 21, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
135 changes: 82 additions & 53 deletions src/scraper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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__)
Expand Down Expand Up @@ -58,80 +58,110 @@ 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[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:
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: 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..."
)
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}")
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"""
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}")
result = self._fetch_with_backoff(f"{self.base_url}/campaigns")
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)"""
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}")
result = self._fetch_with_backoff(f"{self.base_url}/assignments")
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"""
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}")
result = self._fetch_with_backoff(f"{self.base_url}/dispatches")
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"""
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}")
result = self._fetch_with_backoff(f"{self.base_url}/planets")
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"""
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}")
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:
Expand All @@ -140,14 +170,13 @@ 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}")
result = self._fetch_with_backoff(f"{self.base_url}/planet-events")
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)"""
Expand Down