Chemist Warehouse APIchemistwarehouse.co.nz ↗

"""
Chemist Warehouse NZ API Client
A practical example of using the Parse API to search products, browse categories,
get product details, and find store locations.

Get your API key from: https://parse.bot/settings
"""

import os
import requests
from typing import Any, Optional


class ParseClient:
    """Client for interacting with the Chemist Warehouse NZ API via Parse."""

    def __init__(self, api_key: Optional[str] = None):
        """
        Initialize the Parse API client.

        Args:
            api_key: API key for authentication. If not provided, reads from PARSE_API_KEY env var.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "b2758f67-3cb9-4184-ab37-e3262ad57493"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")

        if not self.api_key:
            raise ValueError(
                "API key not provided. Set PARSE_API_KEY environment variable or pass api_key parameter."
            )

    def _call(
        self, endpoint: str, method: str = "POST", **params: Any
    ) -> dict[str, Any]:
        """
        Make an API call to the Parse endpoint.

        Args:
            endpoint: The endpoint name (e.g., 'search_products')
            method: HTTP method ('GET' or 'POST')
            **params: Query/body parameters for the endpoint

        Returns:
            JSON response from the API
        """
        url = f"{self.base_url}/scraper/{self.scraper_id}/{endpoint}"
        headers = {"X-API-Key": self.api_key, "Content-Type": "application/json"}

        if method == "GET":
            response = requests.get(url, headers=headers, params=params)
        elif method == "POST":
            response = requests.post(url, headers=headers, json=params)
        else:
            raise ValueError(f"Unsupported HTTP method: {method}")

        response.raise_for_status()
        return response.json()

    def search_products(
        self,
        query: str,
        page: int = 1,
        size: int = 48,
        sort: Optional[str] = None,
    ) -> dict[str, Any]:
        """
        Search for products by keyword.

        Args:
            query: Search keyword (e.g., 'vitamin c')
            page: Page number (1-based), default 1
            size: Number of items per page, default 48
            sort: Sort option for results

        Returns:
            Dictionary containing items, page, total, and size
        """
        params = {"query": query, "page": page, "size": size}
        if sort:
            params["sort"] = sort
        return self._call("search_products", method="GET", **params)

    def get_category_products(
        self,
        category_id: str,
        page: int = 1,
        size: int = 48,
        sort: Optional[str] = None,
    ) -> dict[str, Any]:
        """
        Get products in a specific category.

        Args:
            category_id: The numeric category ID (e.g., '81' for Vitamins)
            page: Page number (1-based), default 1
            size: Number of items per page, default 48
            sort: Sort option for results

        Returns:
            Dictionary containing items, page, total, and size
        """
        params = {"category_id": category_id, "page": page, "size": size}
        if sort:
            params["sort"] = sort
        return self._call("get_category_products", method="GET", **params)

    def get_product_details(
        self, product_id: str, slug: Optional[str] = None
    ) -> dict[str, Any]:
        """
        Get detailed information for a specific product.

        Args:
            product_id: The numeric product ID (e.g., '96405')
            slug: The product slug (optional)

        Returns:
            Dictionary containing product details including price, description, ingredients, etc.
        """
        params = {"product_id": product_id}
        if slug:
            params["slug"] = slug
        return self._call("get_product_details", method="GET", **params)

    def get_store_locations(
        self,
        search_text: Optional[str] = None,
        lat: Optional[str] = None,
        lng: Optional[str] = None,
    ) -> dict[str, Any]:
        """
        Find Chemist Warehouse store locations.

        Args:
            search_text: Search keyword (e.g., 'Auckland', postcode)
            lat: Latitude for coordinate-based search
            lng: Longitude for coordinate-based search

        Returns:
            Dictionary containing store location items
        """
        params = {}
        if search_text:
            params["search_text"] = search_text
        if lat:
            params["lat"] = lat
        if lng:
            params["lng"] = lng
        return self._call("get_store_locations", method="GET", **params)

    def get_category_list(self) -> dict[str, Any]:
        """
        Get the list of top-level product categories.

        Returns:
            Dictionary containing category items with category_id, name, and url
        """
        return self._call("get_category_list", method="GET")


def main():
    """
    Practical workflow: Search for products, get details for top results,
    explore categories, and find nearby stores.
    """
    # Initialize the client
    client = ParseClient()

    print("=" * 70)
    print("Chemist Warehouse NZ API - Practical Workflow Example")
    print("=" * 70)

    # Step 1: Search for vitamin C products
    print("\n[1] Searching for Vitamin C products...")
    search_results = client.search_products(query="vitamin c", page=1, size=5)
    
    total_found = search_results.get("total", 0)
    items = search_results.get("items", [])
    
    print(f"    ✓ Found {total_found} total products")
    print(f"    ✓ Showing {len(items)} items on page 1\n")

    # Step 2: Collect product IDs and display basic info
    print("[2] Top 3 Vitamin C Products:")
    product_ids = []
    
    for idx, product in enumerate(items[:3], 1):
        product_id = product.get("id")
        name = product.get("name", "Unknown")
        price = product.get("price_cw_nz", "N/A")
        rrp = product.get("rrp_cw_nz", "N/A")
        rating = product.get("bv_star_rating", "N/A")
        votes = product.get("bv_total_votes", "N/A")
        
        product_ids.append(product_id)
        
        print(f"\n    [{idx}] {name}")
        print(f"        Price: ${price} (RRP: ${rrp})")
        print(f"        Rating: {rating}/5 ({votes} votes)")
        print(f"        Product ID: {product_id}")

    # Step 3: Get detailed information for the first product
    if product_ids:
        print(f"\n[3] Getting detailed info for first product (ID: {product_ids[0]})...")
        
        try:
            details = client.get_product_details(product_id=product_ids[0])
            data = details.get("data", {})
            
            print(f"    ✓ Name: {data.get('name', 'N/A')}")
            print(f"    ✓ Price: {data.get('price', 'N/A')}")
            
            description = data.get("description")
            if description:
                desc_preview = description[:120] + "..." if len(description) > 120 else description
                print(f"    ✓ Description: {desc_preview}")
            
            ingredients = data.get("ingredients")
            if ingredients:
                ingredients_preview = ingredients[:100] + "..." if len(ingredients) > 100 else ingredients
                print(f"    ✓ Ingredients: {ingredients_preview}")
            
            directions = data.get("directions")
            if directions:
                directions_preview = directions[:100] + "..." if len(directions) > 100 else directions
                print(f"    ✓ Directions: {directions_preview}")
            
            images = data.get("images", [])
            if images:
                print(f"    ✓ Available {len(images)} product image(s)")
                
        except Exception as e:
            print(f"    ✗ Error fetching details: {e}")

    # Step 4: Get all categories and find Vitamins
    print("\n[4] Fetching product categories...")
    
    try:
        categories_response = client.get_category_list()
        category_items = categories_response.get("data", [])
        
        print(f"    ✓ Found {len(category_items)} categories")
        
        # Find Vitamins category
        vitamins_category_id = None
        for category in category_items:
            if "vitamin" in category.get("name", "").lower():
                vitamins_category_id = category.get("category_id")
                print(f"    ✓ Located: {category.get('name')} (ID: {vitamins_category_id})")
                break
        
        # Step 5: Get products from Vitamins category
        if vitamins_category_id:
            print(f"\n[5] Getting products from Vitamins & Supplements category...")
            category_products = client.get_category_products(
                category_id=vitamins_category_id,
                page=1,
                size=5
            )
            
            total_in_category = category_products.get("total", 0)
            items_in_category = category_products.get("items", [])
            
            print(f"    ✓ Total vitamins available: {total_in_category}")
            print(f"    ✓ Showing {len(items_in_category)} items:\n")
            
            for idx, product in enumerate(items_in_category, 1):
                name = product.get("name", "Unknown")
                price = product.get("price_cw_nz", "N/A")
                rating = product.get("bv_star_rating", "N/A")
                print(f"       {idx}. {name}")
                print(f"          Price: ${price} | Rating: {rating}/5")
                
    except Exception as e:
        print(f"    ✗ Error fetching categories: {e}")

    # Step 6: Find store locations
    print("\n[6] Finding store locations in major cities...")
    
    cities = ["Auckland", "Wellington", "Christchurch"]
    all_stores = []
    
    for city in cities:
        try:
            stores_response = client.get_store_locations(search_text=city)
            store_items = stores_response.get("data", [])
            
            if store_items:
                print(f"    ✓ {city}: Found {len(store_items)} store(s)")
                all_stores.append(store_items[0])  # Keep first store from each city
                
        except Exception as e:
            print(f"    ✗ Error finding stores in {city}: {e}")
    
    # Display store details
    if all_stores:
        print("\n[7] Nearby Store Details:")
        for idx, store in enumerate(all_stores, 1):
            print(f"\n    [{idx}] {store.get('name', 'Unknown Store')}")
            print(f"        Address: {store.get('address', 'N/A')}")
            print(f"        Suburb: {store.get('suburb', 'N/A')} {store.get('postcode', '')}")
            print(f"        Phone: {store.get('phone', 'N/A')}")
            if store.get('email'):
                print(f"        Email: {store.get('email')}")

    print("\n" + "=" * 70)
    print("✓ Workflow completed successfully!")
    print("=" * 70)


if __name__ == "__main__":
    main()