added optional logging to brickset

This commit is contained in:
C committed 2025-10-26 15:47:19 -07:00
1 parent 83769ce7db
commit 9b5d04183d
1 file changed
+26 -4
+26 -4
View File
@@ -17,7 +17,7 @@ dotenv.load_dotenv()
class AsyncBricksetAPI:
"""Asynchronous client for the Brickset API"""
def __init__(self, debug: bool = False):
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")
@@ -25,6 +25,7 @@ class AsyncBricksetAPI:
self.headers = {"UserAgent": "Edderkop/0.1"}
self.throttler = Throttler(rate_limit=500, period=60)
self.user_hash = None
self.logger = logger
async def login(self) -> Dict[str, Any]:
"""Asynchronously authenticate with Brickset API"""
@@ -41,10 +42,15 @@ class AsyncBricksetAPI:
) 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: str) -> Dict[str, Any]:
@@ -73,13 +79,23 @@ class AsyncBricksetAPI:
) 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"):
raise ValueError(f"No data found for set {set_number}")
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]
our_set = {
@@ -111,11 +127,17 @@ class AsyncBricksetAPI:
return our_set
elif response.status == 500:
raise Exception(
error_msg = (
"Internal Server Error: Please check API key and parameters."
)
if self.logger:
self.logger.error(error_msg)
raise Exception(error_msg)
else:
raise Exception(f"Unexpected response: {response.status}")
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"""