#!/usr/bin/env python3 import shopify import traceback import logging from typing import Any, Dict, List # Configure logging logging.basicConfig( level=logging.DEBUG, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) logger = logging.getLogger("shopify_client") class ShopifyClient: def __init__( self, shop_url: str = None, access_token: str = None, # api_key: str = None, # password: str = None, api_version: str = "2023-01", ): """ Initialize the Shopify client with credentials. Args: shop_url: Store domain (e.g., 'your-store.myshopify.com') access_token: Private app token api_key: Public app API key password: Public app password api_version: Shopify API version """ self.shop_url = shop_url self.access_token = access_token # self.api_key = api_key # self.password = password self.api_version = api_version if not self._init_shopify_api(): raise RuntimeError("Failed to initialize Shopify API") def _init_shopify_api(self) -> bool: """Set up Shopify session and test connection.""" if not self.shop_url: logger.error("SHOPIFY_SHOP_URL is required") return False if not (self.access_token or (self.api_key and self.password)): logger.error( "Provide either SHOPIFY_ACCESS_TOKEN or SHOPIFY_API_KEY + SHOPIFY_PASSWORD" ) return False try: logger.debug( f"Connecting to Shopify shop={self.shop_url} version={self.api_version}" ) if self.access_token: session = shopify.Session( self.shop_url, self.api_version, self.access_token ) else: session = shopify.Session(self.shop_url, self.api_version) shopify.ShopifyResource.set_site( f"https://{self.api_key}:{self.password}@{self.shop_url}/admin/api/{self.api_version}" ) shopify.ShopifyResource.activate_session(session) # Test connection shop = shopify.Shop.current() logger.info(f"Connected to Shopify store: {shop.name}") return True except Exception as e: logger.error(f"Error initializing Shopify API: {e}") logger.debug(traceback.format_exc()) return False # =============== PRODUCTS =============== def get_products(self, limit: int = 10) -> List[Dict[str, Any]]: try: products = shopify.Product.find(limit=limit) return [self._product_to_dict(p) for p in products] except Exception as e: logger.error(f"Error retrieving products: {e}") return [] def get_product_details(self, product_id: str) -> Dict[str, Any]: try: product = shopify.Product.find(product_id) return self._product_to_dict(product) except Exception as e: logger.error(f"Error retrieving product details: {e}") return {} # =============== CUSTOMERS =============== def get_customers(self, limit: int = 10) -> List[Dict[str, Any]]: try: customers = shopify.Customer.find(limit=limit) return [self._customer_to_dict(c) for c in customers] except Exception as e: logger.error(f"Error retrieving customers: {e}") return [] def get_customer_details(self, customer_id: str) -> Dict[str, Any]: try: customer = shopify.Customer.find(customer_id) return self._customer_to_dict(customer) except Exception as e: logger.error(f"Error retrieving customer details: {e}") return {} # =============== ORDERS =============== def get_orders(self, limit: int = 10) -> List[Dict[str, Any]]: try: orders = shopify.Order.find(limit=limit) return [self._order_to_dict(o) for o in orders] except Exception as e: logger.error(f"Error retrieving orders: {e}") return [] # =============== SEARCH =============== def search_products(self, query: str, limit: int = 10) -> List[Dict[str, Any]]: try: products = shopify.Product.find(limit=50) query = query.lower() matched = [ self._product_to_dict(p) for p in products if ( query in p.title.lower() or query in p.vendor.lower() or query in p.product_type.lower() or query in p.tags.lower() ) ] return matched[:limit] except Exception as e: logger.error(f"Error searching products: {e}") return [] # =============== STORE INFO =============== def get_store_info(self) -> Dict[str, Any]: try: shop = shopify.Shop.current() return { "id": shop.id, "name": shop.name, "email": shop.email, "domain": shop.domain, "country": shop.country, "currency": shop.currency, "plan_name": shop.plan_name, "shop_owner": shop.shop_owner, } except Exception as e: logger.error(f"Error retrieving store info: {e}") return {} # =============== HELPERS =============== def _product_to_dict(self, product) -> Dict[str, Any]: return { "id": product.id, "title": product.title, "description": product.body_html, "vendor": product.vendor, "product_type": product.product_type, "tags": product.tags, "variants": [ { "id": v.id, "title": v.title, "price": v.price, "sku": v.sku, "inventory_quantity": v.inventory_quantity, } for v in getattr(product, "variants", []) ], "images": [ {"id": img.id, "src": img.src, "position": img.position} for img in getattr(product, "images", []) ], } def _customer_to_dict(self, customer) -> Dict[str, Any]: return { "id": customer.id, "email": customer.email, "first_name": customer.first_name, "last_name": customer.last_name, "orders_count": customer.orders_count, "total_spent": customer.total_spent, "created_at": str(customer.created_at), "addresses": [ { "address1": a.address1, "city": a.city, "province": a.province, "country": a.country, "zip": a.zip, } for a in getattr(customer, "addresses", []) ], } def _order_to_dict(self, order) -> Dict[str, Any]: return { "id": order.id, "order_number": order.order_number, "email": order.email, "created_at": str(order.created_at), "total_price": order.total_price, "subtotal_price": order.subtotal_price, "total_tax": order.total_tax, "currency": order.currency, "financial_status": order.financial_status, "fulfillment_status": order.fulfillment_status, "customer": ( self._customer_to_dict(order.customer) if getattr(order, "customer", None) else {} ), "shipping_address": ( { "name": order.shipping_address.name, "address1": order.shipping_address.address1, "city": order.shipping_address.city, "province": order.shipping_address.province, "country": order.shipping_address.country, "zip": order.shipping_address.zip, } if getattr(order, "shipping_address", None) else {} ), "line_items": [ { "id": li.id, "title": li.title, "quantity": li.quantity, "price": li.price, "sku": li.sku, "product_id": li.product_id, "variant_id": li.variant_id, } for li in getattr(order, "line_items", []) ], } # # Example usage # if __name__ == "__main__": # client = ShopifyClient() # print(client.get_store_info()) # print(client.get_products(limit=5))