""" abricksetapi.py - Asynchronous Brickset API Client This module provides asynchronous interactions with the Brickset API. """ import asyncio import os import aiohttp import json import math import dotenv from typing import Dict, Any, Optional from datetime import datetime, timedelta from . import throttler Throttler = throttler.GlobalThrottler dotenv.load_dotenv() class AsyncBricksetAPI: """Asynchronous client for the Brickset API""" def __init__(self, debug: bool = False, logger=None): self.username = os.getenv("BRICKSET_USERNAME") self.password = os.getenv("BRICKSET_PASS") self.api_key = os.getenv("BRICKSET_API_KEY") self.debug = debug self.headers = {"UserAgent": "studspider/0.1"} self.throttler = Throttler(rate_limit=500, period=60) self.user_hash = None self.logger = logger # API usage tracking self.daily_requests_used: Optional[int] = None self.daily_requests_limit = 1000 self.usage_last_checked: Optional[datetime] = None self.usage_check_interval = timedelta(hours=1) # Check usage every hour async def get_key_usage_stats(self) -> Dict[str, Any]: """ Fetch API key usage statistics from Brickset. Returns: dict: Usage stats including dateStamp and apiCalls """ url = "https://brickset.com/api/v3.asmx/getKeyUsageStats" params = {"apiKey": self.api_key} async with aiohttp.ClientSession() as session: async with session.get( url, params=params, headers=self.headers ) as response: if response.status == 200: data = await response.read() result = json.loads(data.decode("utf-8")) if self.debug: print(f"[@DEBUG] API usage stats: {result}") if self.logger: self.logger.debug(f"API usage stats: {result}") return result else: error_msg = f"Failed to get usage stats: HTTP {response.status}" if self.logger: self.logger.error(error_msg) raise Exception(error_msg) async def check_and_update_usage(self) -> bool: """ Check current API usage and update internal counter. Automatically called before each request if needed. Returns: bool: True if under limit, False if at/over limit """ now = datetime.now() # Check usage if we haven't checked yet or if it's been over an hour should_check = ( self.daily_requests_used is None or self.usage_last_checked is None or (now - self.usage_last_checked) > self.usage_check_interval ) if should_check: try: stats = await self.get_key_usage_stats() # Brickset returns {"status": "success", "apiKeyUsage": [...]} usage_array = stats.get("apiKeyUsage", []) if usage_array and len(usage_array) > 0: today_stats = usage_array[0] # Most recent day self.daily_requests_used = today_stats.get("count", 0) self.usage_last_checked = now if self.logger: self.logger.info( f"API usage: {self.daily_requests_used}/{self.daily_requests_limit}" ) if self.debug: print( f"[@DEBUG] API usage: {self.daily_requests_used}/{self.daily_requests_limit}" ) except Exception as e: if self.logger: self.logger.warning(f"Failed to check API usage: {e}") # Don't block requests if usage check fails return True # Check if we're over the limit if self.daily_requests_used is not None: if self.daily_requests_used >= self.daily_requests_limit: error_msg = f"Daily API limit reached: {self.daily_requests_used}/{self.daily_requests_limit}" if self.logger: self.logger.error(error_msg) if self.debug: print(f"[@ERROR] {error_msg}") return False # Warn when approaching limit if self.daily_requests_used >= self.daily_requests_limit * 0.9: warning_msg = f"Approaching daily API limit: {self.daily_requests_used}/{self.daily_requests_limit}" if self.logger: self.logger.warning(warning_msg) if self.debug: print(f"[@WARNING] {warning_msg}") return True async def login(self) -> Dict[str, Any]: """Asynchronously authenticate with Brickset API""" url = "https://brickset.com/api/v3.asmx/login" params = { "username": self.username, "password": self.password, "apiKey": self.api_key, } async with aiohttp.ClientSession() as session: async with session.get( url, params=params, headers=self.headers # type: ignore ) as response: if self.debug: print(f"[@DEBUG] Login response status: {response.status}") if self.logger: self.logger.debug(f"Login response status: {response.status}") data = await response.read() result = json.loads(data.decode("utf-8")) self.user_hash = result.get("hash") if self.logger: self.logger.info("Successfully authenticated with Brickset API") return result async def get_set(self, set_number: int) -> Dict[str, Any]: """ Fetch LEGO set information asynchronously Args: set_number (str): LEGO set number Returns: Dict containing set information Raises: Exception: If daily API limit exceeded or API error """ # Check API usage before making request if not await self.check_and_update_usage(): raise Exception( f"Brickset API daily limit reached ({self.daily_requests_limit} requests/day). " "Please try again tomorrow." ) if not self.user_hash: await self.login() url = "https://brickset.com/api/v3.asmx/getSets" params = { "apiKey": self.api_key, "userHash": self.user_hash, "params": json.dumps({"setNumber": f"{set_number}-1"}), } print(f"[@DEBUG] {params}...") async with aiohttp.ClientSession() as session: async with session.get( url, params=params, headers=self.headers ) as response: if self.debug: print(f"[@DEBUG] Get set response status: {response.status}") if self.logger: self.logger.debug(f"Get set response status: {response.status}") if response.status == 200: data = await response.read() result = json.loads(data.decode("utf-8")) if not result.get("sets"): error_msg = f"No data found for set {set_number}" if self.logger: self.logger.error(error_msg) raise ValueError(error_msg) if self.logger: self.logger.info( f"Successfully retrieved data for set {set_number}" ) set_data = result["sets"][0] # Try to get UPC barcode, fall back to EAN if not available barcode_data = set_data.get("barcode", {}) upc_code = barcode_data.get("UPC") or barcode_data.get("EAN") # Get retail price before rounding retail_price = set_data["LEGOCom"]["US"]["retailPrice"] our_set = { "set_num": set_data["number"], "set_name": set_data["name"], "set_theme": set_data["theme"], "set_image": set_data["image"]["thumbnailURL"], "set_barcode": upc_code, "set_retailprice": round(retail_price), } # Calculate adjusted price (10% markup), then round up adjusted_price = retail_price * 1.10 our_set["set_adjustredprice"] = math.ceil(adjusted_price) # Set PW inventory name with theme our_set["pw_name"] = ( f"{our_set['set_num']} - {our_set['set_theme'].upper()} - {our_set['set_name']}" ) # Flag for manual price review if needed if our_set["set_retailprice"] >= 110: our_set["requires_manual_price"] = True return our_set elif response.status == 500: error_msg = ( "Internal Server Error: Please check API key and parameters." ) if self.logger: self.logger.error(error_msg) raise Exception(error_msg) else: error_msg = f"Unexpected response: {response.status}" if self.logger: self.logger.error(error_msg) raise Exception(error_msg) async def throttled_get_set(self, set_number: str) -> Dict[str, Any]: """Get set information with rate limiting""" async with self.throttler: return await self.get_set(set_number) async def main(): """Example usage""" client = AsyncBricksetAPI(debug=True) try: set_info = await client.throttled_get_set("10349") print(json.dumps(set_info, indent=2)) except Exception as e: print(f"Error: {e}") if __name__ == "__main__": asyncio.run(main())