added optional logging to brickset
This commit is contained in:
1 parent
41ce94c85c
commit
32664c3a10
1 file changed
+26
-4
+26
-4
@@ -17,7 +17,7 @@ dotenv.load_dotenv()
|
|||||||
class AsyncBricksetAPI:
|
class AsyncBricksetAPI:
|
||||||
"""Asynchronous client for the Brickset API"""
|
"""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.username = os.getenv("BRICKSET_USERNAME")
|
||||||
self.password = os.getenv("BRICKSET_PASS")
|
self.password = os.getenv("BRICKSET_PASS")
|
||||||
self.api_key = os.getenv("BRICKSET_API_KEY")
|
self.api_key = os.getenv("BRICKSET_API_KEY")
|
||||||
@@ -25,6 +25,7 @@ class AsyncBricksetAPI:
|
|||||||
self.headers = {"UserAgent": "Edderkop/0.1"}
|
self.headers = {"UserAgent": "Edderkop/0.1"}
|
||||||
self.throttler = Throttler(rate_limit=500, period=60)
|
self.throttler = Throttler(rate_limit=500, period=60)
|
||||||
self.user_hash = None
|
self.user_hash = None
|
||||||
|
self.logger = logger
|
||||||
|
|
||||||
async def login(self) -> Dict[str, Any]:
|
async def login(self) -> Dict[str, Any]:
|
||||||
"""Asynchronously authenticate with Brickset API"""
|
"""Asynchronously authenticate with Brickset API"""
|
||||||
@@ -41,10 +42,15 @@ class AsyncBricksetAPI:
|
|||||||
) as response:
|
) as response:
|
||||||
if self.debug:
|
if self.debug:
|
||||||
print(f"[@DEBUG] Login response status: {response.status}")
|
print(f"[@DEBUG] Login response status: {response.status}")
|
||||||
|
if self.logger:
|
||||||
|
self.logger.debug(f"Login response status: {response.status}")
|
||||||
|
|
||||||
data = await response.read()
|
data = await response.read()
|
||||||
result = json.loads(data.decode("utf-8"))
|
result = json.loads(data.decode("utf-8"))
|
||||||
self.user_hash = result.get("hash")
|
self.user_hash = result.get("hash")
|
||||||
|
|
||||||
|
if self.logger:
|
||||||
|
self.logger.info("Successfully authenticated with Brickset API")
|
||||||
return result
|
return result
|
||||||
|
|
||||||
async def get_set(self, set_number: str) -> Dict[str, Any]:
|
async def get_set(self, set_number: str) -> Dict[str, Any]:
|
||||||
@@ -73,13 +79,23 @@ class AsyncBricksetAPI:
|
|||||||
) as response:
|
) as response:
|
||||||
if self.debug:
|
if self.debug:
|
||||||
print(f"[@DEBUG] Get set response status: {response.status}")
|
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:
|
if response.status == 200:
|
||||||
data = await response.read()
|
data = await response.read()
|
||||||
result = json.loads(data.decode("utf-8"))
|
result = json.loads(data.decode("utf-8"))
|
||||||
|
|
||||||
if not result.get("sets"):
|
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]
|
set_data = result["sets"][0]
|
||||||
our_set = {
|
our_set = {
|
||||||
@@ -111,11 +127,17 @@ class AsyncBricksetAPI:
|
|||||||
return our_set
|
return our_set
|
||||||
|
|
||||||
elif response.status == 500:
|
elif response.status == 500:
|
||||||
raise Exception(
|
error_msg = (
|
||||||
"Internal Server Error: Please check API key and parameters."
|
"Internal Server Error: Please check API key and parameters."
|
||||||
)
|
)
|
||||||
|
if self.logger:
|
||||||
|
self.logger.error(error_msg)
|
||||||
|
raise Exception(error_msg)
|
||||||
else:
|
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]:
|
async def throttled_get_set(self, set_number: str) -> Dict[str, Any]:
|
||||||
"""Get set information with rate limiting"""
|
"""Get set information with rate limiting"""
|
||||||
|
|||||||
Reference in new issue
Block a user