testing subclassed asquare
This commit is contained in:
C committed
1 parent
a419dd498f
commit
369ccdbfc0
1 file changed
+218
@@ -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())
|
||||
Reference in new issue
Block a user