Nabis APInabis.com ↗

"""
Nabis Marketplace API Client

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

import os
import requests
from typing import Optional, Dict, Any, List


class ParseClient:
    """Client for interacting with the Nabis Marketplace API through Parse."""

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

        Args:
            api_key: Parse API key. If not provided, reads from PARSE_API_KEY env var.
        """
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "a82d4854-b40f-40a1-8073-71cb657b931e"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")

        if not self.api_key:
            raise ValueError("API key required. Pass as argument or set PARSE_API_KEY env var.")

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

        Args:
            endpoint: The endpoint name (e.g., 'get_inventory_ca')
            method: HTTP method ('GET' or 'POST')
            **params: Parameters to send with the request

        Returns:
            Response JSON as dictionary

        Raises:
            requests.RequestException: If the API call fails
        """
        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 get_inventory_ca(
        self,
        access_token: str,
        limit: int = 500,
        page: int = 1
    ) -> Dict[str, Any]:
        """
        Retrieve paginated list of cannabis inventory/products available in California.

        Args:
            access_token: Nabis x-nabis-access-token header value
            limit: Max results per page (1-500, default 500)
            page: Page number (default 1)

        Returns:
            Dictionary with 'items' and 'total' keys
        """
        return self._call(
            "get_inventory_ca",
            method="GET",
            access_token=access_token,
            limit=limit,
            page=page
        )

    def get_inventory_by_sku_ca(
        self,
        access_token: str,
        sku_code: str
    ) -> Dict[str, Any]:
        """
        Retrieve detailed inventory/product information for a specific SKU in California.

        Args:
            access_token: Nabis x-nabis-access-token header value
            sku_code: SKU code of the product

        Returns:
            Dictionary with product details including skuCode and skuName
        """
        return self._call(
            "get_inventory_by_sku_ca",
            method="GET",
            access_token=access_token,
            sku_code=sku_code
        )

    def get_inventory_ny(
        self,
        access_token: str,
        limit: int = 500,
        page: int = 1
    ) -> Dict[str, Any]:
        """
        Retrieve paginated list of cannabis inventory/products available in New York.

        Args:
            access_token: Nabis x-nabis-access-token header value
            limit: Max results per page (1-500, default 500)
            page: Page number (default 1)

        Returns:
            Dictionary with 'items' and 'total' keys
        """
        return self._call(
            "get_inventory_ny",
            method="GET",
            access_token=access_token,
            limit=limit,
            page=page
        )

    def get_inventory_by_product_code_ny(
        self,
        access_token: str,
        item_code: str
    ) -> Dict[str, Any]:
        """
        Retrieve detailed inventory/product information for a specific product in New York.

        Args:
            access_token: Nabis x-nabis-access-token header value
            item_code: Item code of the product in NY

        Returns:
            Dictionary with product details including itemCode and skuName
        """
        return self._call(
            "get_inventory_by_product_code_ny",
            method="GET",
            access_token=access_token,
            item_code=item_code
        )

    def get_retailers(
        self,
        access_token: str,
        limit: int = 500,
        page: int = 1
    ) -> Dict[str, Any]:
        """
        Retrieve list of licensed dispensary retailers on the Nabis platform.

        Args:
            access_token: Nabis x-nabis-access-token header value
            limit: Max results per page (default 500)
            page: Page number (default 1)

        Returns:
            Dictionary with 'items' array of retailers
        """
        return self._call(
            "get_retailers",
            method="GET",
            access_token=access_token,
            limit=limit,
            page=page
        )

    def get_orders(
        self,
        access_token: str,
        limit: int = 500,
        page: int = 1
    ) -> Dict[str, Any]:
        """
        Retrieve paginated list of wholesale orders placed through the Nabis platform.

        Args:
            access_token: Nabis x-nabis-access-token header value
            limit: Max results per page (default 500)
            page: Page number (default 1)

        Returns:
            Dictionary with 'items' array of orders
        """
        return self._call(
            "get_orders",
            method="GET",
            access_token=access_token,
            limit=limit,
            page=page
        )

    def get_warehouses(
        self,
        access_token: str,
        limit: int = 500,
        page: int = 1
    ) -> Dict[str, Any]:
        """
        Retrieve list of Nabis warehouse locations.

        Args:
            access_token: Nabis x-nabis-access-token header value
            limit: Max results per page (default 500)
            page: Page number (default 1)

        Returns:
            Dictionary with 'items' array of warehouses
        """
        return self._call(
            "get_warehouses",
            method="GET",
            access_token=access_token,
            limit=limit,
            page=page
        )

    def get_invoices(
        self,
        access_token: str,
        limit: int = 500,
        page: int = 1
    ) -> Dict[str, Any]:
        """
        Retrieve paginated list of invoices/orders.

        Args:
            access_token: Nabis x-nabis-access-token header value
            limit: Max results per page (default 500)
            page: Page number (default 1)

        Returns:
            Dictionary with 'items' array of invoices
        """
        return self._call(
            "get_invoices",
            method="GET",
            access_token=access_token,
            limit=limit,
            page=page
        )

    def get_nabis_days_off(
        self,
        access_token: str
    ) -> Dict[str, Any]:
        """
        Retrieve list of Nabis non-delivery/holiday dates.

        Args:
            access_token: Nabis x-nabis-access-token header value

        Returns:
            Dictionary with 'items' array of days off with date and description
        """
        return self._call(
            "get_nabis_days_off",
            method="GET",
            access_token=access_token
        )


if __name__ == "__main__":
    # Example usage: Realistic workflow for a cannabis distributor
    # This demonstrates a practical use case: finding retailers, checking their orders,
    # and looking up product details for items in those orders.

    # Initialize client
    client = ParseClient()

    # Use a sample access token (replace with actual token)
    access_token = "your_nabis_access_token_here"

    print("=" * 60)
    print("NABIS MARKETPLACE WORKFLOW EXAMPLE")
    print("=" * 60)

    # Step 1: Get all retailers on the platform
    print("\n📍 Step 1: Fetching all retailers...")
    retailers_response = client.get_retailers(access_token=access_token, limit=10)
    retailers = retailers_response.get("items", [])
    print(f"Found {len(retailers)} retailers")

    if retailers:
        for retailer in retailers[:3]:  # Show first 3
            retailer_id = retailer.get("id")
            retailer_name = retailer.get("name", "Unknown")
            print(f"  • {retailer_name} (ID: {retailer_id})")

    # Step 2: Get available inventory in California
    print("\n📦 Step 2: Fetching California inventory...")
    ca_inventory_response = client.get_inventory_ca(
        access_token=access_token,
        limit=5
    )
    ca_products = ca_inventory_response.get("items", [])
    total_ca = ca_inventory_response.get("total", 0)
    print(f"Total CA products available: {total_ca}")

    # Step 3: Get detailed info for each product
    if ca_products:
        print("\n🔍 Step 3: Getting detailed product information...")
        for product in ca_products:
            sku_code = product.get("skuCode")
            sku_name = product.get("skuName", "Unknown")

            if sku_code:
                try:
                    product_details = client.get_inventory_by_sku_ca(
                        access_token=access_token,
                        sku_code=sku_code
                    )
                    detailed_name = product_details.get("skuName", sku_name)
                    print(f"  • {detailed_name} (SKU: {sku_code})")
                except Exception as e:
                    print(f"  • {sku_name} (SKU: {sku_code}) - Details unavailable")

    # Step 4: Check recent orders
    print("\n📋 Step 4: Fetching recent orders...")
    orders_response = client.get_orders(access_token=access_token, limit=5)
    orders = orders_response.get("items", [])
    print(f"Retrieved {len(orders)} recent orders")

    total_order_value = 0
    for order in orders:
        order_id = order.get("id")
        total_price = order.get("totalPrice", 0)
        total_order_value += total_price
        print(f"  • Order #{order_id}: ${total_price:,.2f}")

    if orders:
        print(f"  📊 Total order value: ${total_order_value:,.2f}")

    # Step 5: Check warehouse locations
    print("\n🏭 Step 5: Fetching warehouse locations...")
    warehouses_response = client.get_warehouses(access_token=access_token)
    warehouses = warehouses_response.get("items", [])
    print(f"Nabis warehouses:")
    for warehouse in warehouses:
        warehouse_name = warehouse.get("name", "Unknown")
        warehouse_id = warehouse.get("id")
        print(f"  • {warehouse_name} (ID: {warehouse_id})")

    # Step 6: Check days off (holidays when orders can't be placed)
    print("\n🗓️  Step 6: Checking Nabis days off...")
    days_off_response = client.get_nabis_days_off(access_token=access_token)
    days_off = days_off_response.get("items", [])

    if days_off:
        print("Nabis non-delivery dates:")
        for day_off in days_off[:5]:  # Show first 5
            date = day_off.get("date", "Unknown")
            description = day_off.get("description", "Holiday")
            print(f"  • {date}: {description}")

    # Step 7: Get invoices (billing records)
    print("\n💳 Step 7: Fetching recent invoices...")
    invoices_response = client.get_invoices(access_token=access_token, limit=5)
    invoices = invoices_response.get("items", [])
    print(f"Retrieved {len(invoices)} recent invoices")

    for invoice in invoices:
        invoice_id = invoice.get("id")
        invoice_number = invoice.get("invoiceNumber")
        print(f"  • Invoice #{invoice_number} (ID: {invoice_id})")

    print("\n" + "=" * 60)
    print("✅ Workflow complete!")
    print("=" * 60)