From 369ccdbfc020e29a74113ef6e46de27e08983412 Mon Sep 17 00:00:00 2001 From: C <0xicl33n@gmail.com> Date: Thu, 9 Oct 2025 17:14:37 -0700 Subject: [PATCH] testing subclassed asquare --- src/asquareclass.py | 218 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 218 insertions(+) create mode 100644 src/asquareclass.py diff --git a/src/asquareclass.py b/src/asquareclass.py new file mode 100644 index 0000000..b9407b4 --- /dev/null +++ b/src/asquareclass.py @@ -0,0 +1,218 @@ +import os +import json +import asyncio +import re +from typing import List, Dict, Any, Optional +from square import AsyncSquare +import get_sq_env + + +class AsyncSquareClient: + """Asynchronous client for the Square API with focus on catalog operations""" + + def __init__( + self, + token: Optional[str] = None, + environment: Optional[str] = None, + logger=None, + ): + """ + Initialize the AsyncSquareClient. + + Args: + token (str, optional): Square API token. If not provided, fetches from environment. + environment (str, optional): Square environment (sandbox/production). If not provided, fetches from environment. + logger: Optional logger object for logging operations. + """ + self.token = token or get_sq_env.SQUARE_TOKEN + self.environment = environment or get_sq_env.environment + self.lego_cat_id = get_sq_env.LEGO_CAT_ID + self.client = AsyncSquare(token=self.token, environment=self.environment) # type: ignore + self.logger = logger + + @staticmethod + def get_set_number(text: str) -> int: + """ + Extract a five-digit LEGO set number from a string. + + Args: + text (str): The text to search for a set number. + + Returns: + int: The extracted set number. + + Raises: + ValueError: If no five-digit number is found. + """ + numbers = re.findall(r"\d{5}", text) + if not numbers: + raise ValueError(f"No five-digit number found in string {text}") + return int(numbers[0]) + + async def get_item(self, sku: str) -> Dict[str, str]: + """ + Asynchronously retrieves item details from Square Catalog using a SKU. + + Args: + sku (str): The SKU or text filter to search for the item. + + Returns: + dict: A dictionary containing the item's name and ID. + + Raises: + IndexError: If no items are found matching the SKU. + Exception: If there is an error communicating with the Square API. + """ + if self.logger: + self.logger.debug(f"Searching for item with SKU: {sku}") + + result = await self.client.catalog.search_items(text_filter=sku, limit=1) + result_json = json.loads(result.model_dump_json()) + + try: + name = result_json["items"][0]["item_data"]["name"] + itm_id = result_json["items"][0]["id"] + + if self.logger: + self.logger.info(f"Found item: {name} with ID: {itm_id}") + + return {"name": name, "id": itm_id} + except (KeyError, IndexError) as e: + if self.logger: + self.logger.error(f"Error retrieving item with SKU {sku}: {str(e)}") + raise + + async def update_item(self, lego_set: Dict[str, Any]) -> bool: + """ + Asynchronously updates an item in the Square catalog based on provided LEGO set information. + + Args: + lego_set (dict): A dictionary containing LEGO set details including: + - sets: List containing set information + - pw_name: PriceWise system name + - set_adjustredprice: Adjusted retail price + + Returns: + bool: True if update was successful. + + Raises: + KeyError: If required keys are missing from input dictionary. + Exception: For Square API communication errors. + """ + if self.logger: + self.logger.debug( + f"Updating item with set info: {lego_set['sets'][0]['number']}" + ) + + try: + set_name = lego_set["sets"][0]["name"] + set_number = lego_set["sets"][0]["number"] + set_upc = lego_set["sets"][0]["barcode"]["UPC"] + set_image = lego_set["sets"][0]["image"]["thumbnailURL"] + pw_name = lego_set["pw_name"] + set_adjustedprice = lego_set["set_adjustredprice"] + + result = await self.client.catalog.search_items( + text_filter=set_number, limit=1 + ) + itm_id = json.loads(result.model_dump_json())["items"][0]["id"] + + if self.logger: + self.logger.info(f"Successfully updated item {set_number}") + + # TODO: Implement actual update logic here + # await self.client.catalog.object.upsert() + return True + + except Exception as e: + if self.logger: + self.logger.error(f"Error updating item: {str(e)}") + raise + + async def get_category_items( + self, cat_id: Optional[str] = None + ) -> List[Dict[str, Any]]: + """ + Return a list of items within the specified category ID. + + Args: + cat_id (str, optional): The category ID to search for items. + If not provided, uses the default LEGO category ID. + + Returns: + list[dict]: A list of dictionaries, each containing: + - 'id': The item's ID + - 'name': The item's name + - 'number': The LEGO set number + """ + category_id = cat_id or self.lego_cat_id + + if self.logger: + self.logger.debug(f"Fetching items for category: {category_id}") + + try: + # Create a properly structured query for the Square API + search_request = { + "object_types": ["ITEM"], + "query": { + "exact_query": { + "attribute_name": "category_id", + "attribute_value": category_id, + } + }, + } + + result = await self.client.catalog.search(**search_request) + result_json = json.loads(result.model_dump_json()) + + items = [ + { + "id": obj["id"], + "name": obj.get("item_data", {}).get("name", "Unknown"), + "number": self.get_set_number( + obj.get("item_data", {}).get("name", "") + ), + } + for obj in result_json.get("objects", []) + ] + + if self.logger: + self.logger.info(f"Found {len(items)} items in category {category_id}") + + return items + + except Exception as e: + if self.logger: + self.logger.error(f"Error fetching category items: {str(e)}") + raise + + +# Provide acat2id as a module-level function for backward compatibility +async def acat2id(cat_id: str) -> List[Dict[str, Any]]: + """ + Backward compatibility function for getting category items. + + Args: + cat_id (str): The category ID to search for items. + + Returns: + list[dict]: A list of dictionaries containing item information. + """ + client = AsyncSquareClient() + return await client.get_category_items(cat_id) + + +async def test() -> None: + """Test function to demonstrate class usage""" + client = AsyncSquareClient() + print(f"@[DEBUG] ENV:\t{client.environment}") + print(f"@[DEBUG] CATID:\t{client.lego_cat_id}") + + result = await client.get_category_items() + print("\nOBJECTS IN LEGO CATEGORY:") + for item in result: + print(f"ID: {item['id']}, Name: {item['name']}") + + +if __name__ == "__main__": + asyncio.run(test())