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())