import all

src/abricksetapi.py

@@ -0,0 +1,137 @@
+"""
+abricksetapi.py - Asynchronous Brickset API Client
+This module provides asynchronous interactions with the Brickset API.
+"""
+
+import asyncio
+import os
+import aiohttp
+import json
+import dotenv
+from typing import Dict, Any
+from asyncio_throttle.throttler import Throttler
+
+dotenv.load_dotenv()
+
+
+class AsyncBricksetAPI:
+    """Asynchronous client for the Brickset API"""
+
+    def __init__(self, debug: bool = False):
+        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": "Edderkop/0.1"}
+        self.throttler = Throttler(rate_limit=500, period=60)
+        self.user_hash = None
+
+    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}")
+
+                data = await response.read()
+                result = json.loads(data.decode("utf-8"))
+                self.user_hash = result.get("hash")
+                return result
+
+    async def get_set(self, set_number: str) -> Dict[str, Any]:
+        """
+        Fetch LEGO set information asynchronously
+
+        Args:
+            set_number (str): LEGO set number
+
+        Returns:
+            Dict containing set information
+        """
+        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"}),
+        }
+
+        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 response.status == 200:
+                    data = await response.read()
+                    result = json.loads(data.decode("utf-8"))
+
+                    if not result.get("sets"):
+                        raise ValueError(f"No data found for set {set_number}")
+
+                    set_data = result["sets"][0]
+                    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": set_data["barcode"]["UPC"],
+                        "set_retailprice": round(
+                            set_data["LEGOCom"]["US"]["retailPrice"]
+                        ),
+                    }
+
+                    # Calculate adjusted price (10% markup)
+                    retail_price = our_set["set_retailprice"]
+                    our_set["set_adjustredprice"] = retail_price + (
+                        (10 / 100) * retail_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:
+                    raise Exception(
+                        "Internal Server Error: Please check API key and parameters."
+                    )
+                else:
+                    raise Exception(f"Unexpected response: {response.status}")
+
+    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())