Octopart APIoctopart.com ↗

"""
Octopart API - Electronic Parts Search and Details

Search and retrieve detailed data for electronic parts from Octopart, including pricing,
stock, and specifications across multiple distributors.

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 Parse Octopart API."""

    def __init__(self, api_key: Optional[str] = None):
        """Initialize the Parse client with API credentials."""
        self.base_url = "https://api.parse.bot"
        self.scraper_id = "29bd1b9e-8808-4396-9ded-37aba5aa0ae4"
        self.api_key = api_key or os.getenv("PARSE_API_KEY")

        if not self.api_key:
            raise ValueError("API key must be provided or set in PARSE_API_KEY environment variable")

    def _call(self, endpoint: str, method: str = "POST", **params) -> dict[str, Any]:
        """Make an API call to the Parse endpoint."""
        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, timeout=30)
        elif method == "POST":
            response = requests.post(url, headers=headers, json=params, timeout=30)
        else:
            raise ValueError(f"Unsupported HTTP method: {method}")

        response.raise_for_status()
        return response.json()

    def search_parts(
        self,
        query: str,
        limit: int = 10,
        offset: int = 0,
        currency: str = "USD",
        country: str = "US",
        in_stock_only: bool = False
    ) -> dict[str, Any]:
        """
        Search for electronic parts by keyword with pricing and availability.

        Args:
            query: Search keyword (e.g., 'resistor', 'capacitor', 'Arduino')
            limit: Maximum number of results to return (default: 10)
            offset: Number of results to skip (default: 0)
            currency: Currency code for pricing (default: 'USD')
            country: Country code for pricing and availability (default: 'US')
            in_stock_only: Filter to only show in-stock parts (default: False)

        Returns:
            Dictionary containing search results and total hits
        """
        return self._call(
            "search_parts",
            method="GET",
            query=query,
            limit=limit,
            offset=offset,
            currency=currency,
            country=country,
            in_stock_only=in_stock_only
        )

    def get_part_detail(
        self,
        part_id: Optional[str] = None,
        mpn: Optional[str] = None,
        manufacturer: Optional[str] = None,
        currency: str = "USD",
        country: str = "US"
    ) -> dict[str, Any]:
        """
        Retrieve detailed information for a specific part by its ID or MPN.

        Args:
            part_id: Octopart part ID
            mpn: Manufacturer Part Number
            manufacturer: Manufacturer name (optional, used when searching by MPN)
            currency: Currency code (default: 'USD')
            country: Country code (default: 'US')

        Returns:
            Dictionary containing part details including specs and sellers
        """
        params = {
            "currency": currency,
            "country": country
        }

        if part_id:
            params["part_id"] = part_id
        if mpn:
            params["mpn"] = mpn
        if manufacturer:
            params["manufacturer"] = manufacturer

        return self._call("get_part_detail", method="GET", **params)

    def get_categories(self) -> dict[str, Any]:
        """
        Retrieve the electronic parts category tree.

        Returns:
            Dictionary containing hierarchical categories
        """
        return self._call("get_categories", method="GET")

    def get_manufacturers(self) -> dict[str, Any]:
        """
        Retrieve a list of electronic part manufacturers.

        Returns:
            Dictionary containing manufacturers with IDs and aliases
        """
        return self._call("get_manufacturers", method="GET")


def format_price(price: float, currency: str = "USD") -> str:
    """Format price with currency symbol."""
    symbols = {"USD": "$", "EUR": "€", "GBP": "£"}
    symbol = symbols.get(currency, currency)
    return f"{symbol}{price:.3f}"


def main():
    """Demonstrate a practical workflow with the Octopart API."""
    # Initialize the client
    client = ParseClient()

    print("=" * 80)
    print("OCTOPART API - ELECTRONIC PARTS SEARCH WORKFLOW")
    print("=" * 80)

    # Step 1: Search for capacitors with stock filter
    print("\n[STEP 1] Searching for 'capacitor' (in stock only)...")
    search_results = client.search_parts(
        query="capacitor",
        limit=5,
        currency="USD",
        country="US",
        in_stock_only=True
    )

    total_hits = search_results.get("data", {}).get("total_hits", 0)
    results = search_results.get("data", {}).get("results", [])
    
    print(f"✓ Found {total_hits:,} matching parts")
    print(f"✓ Displaying {len(results)} results\n")

    part_ids = []
    for i, part in enumerate(results, 1):
        part_ids.append(part.get("id"))
        print(f"  [{i}] {part.get('name', 'Unknown')}")
        print(f"      MPN: {part.get('mpn', 'N/A')}")
        print(f"      Manufacturer: {part.get('manufacturer', 'N/A')}")
        print(f"      Description: {part.get('description', 'No description')[:70]}")
        
        # Show quick pricing from sellers
        sellers = part.get("sellers", [])
        if sellers:
            seller = sellers[0]
            seller_name = seller.get("name", "Unknown")
            offers = seller.get("offers", [])
            if offers:
                offer = offers[0]
                inventory = offer.get("inventory_level", 0)
                prices = offer.get("prices", [])
                if prices:
                    price = prices[0].get("converted_price", 0)
                    currency = prices[0].get("converted_currency", "USD")
                    print(f"      {seller_name}: {inventory} units @ {format_price(price, currency)}")
        print()

    # Step 2: Get detailed information for the first part
    if part_ids:
        print("[STEP 2] Fetching detailed specs for first part...")
        first_part_id = part_ids[0]
        
        detail = client.get_part_detail(part_id=first_part_id)
        part_data = detail.get("data", {})
        
        print(f"✓ Part ID: {part_data.get('id')}")
        print(f"  MPN: {part_data.get('mpn')}")
        print(f"  Manufacturer: {part_data.get('manufacturer', {}).get('name', 'N/A')}")
        
        # Display specifications
        specs = part_data.get("specs", [])
        if specs:
            print(f"\n  Specifications:")
            for spec in specs[:5]:  # Show first 5 specs
                attr = spec.get("attribute", {})
                attr_name = attr.get("name", "Unknown")
                value = spec.get("display_value", "N/A")
                print(f"    • {attr_name}: {value}")
        
        # Display seller information with pricing tiers
        print(f"\n  Sellers & Pricing:")
        sellers = part_data.get("sellers", [])
        for seller in sellers[:3]:  # Show top 3 sellers
            company = seller.get("company", {})
            seller_name = company.get("name", "Unknown")
            is_auth = "✓" if seller.get("is_authorized") else "✗"
            
            print(f"\n    {seller_name} (Authorized: {is_auth})")
            
            offers = seller.get("offers", [])
            for offer in offers[:1]:  # Show first offer per seller
                inventory = offer.get("inventory_level", 0)
                moq = offer.get("moq", 1)
                sku = offer.get("sku", "N/A")
                
                print(f"      SKU: {sku}")
                print(f"      Stock: {inventory} units | MOQ: {moq}")
                
                prices = offer.get("prices", [])
                for price_tier in prices[:3]:  # Show first 3 price tiers
                    price = price_tier.get("converted_price", 0)
                    quantity = price_tier.get("quantity", 0)
                    currency = price_tier.get("converted_currency", "USD")
                    print(f"        Qty {quantity}: {format_price(price, currency)}/unit")

    # Step 3: Search for resistors in different currency
    print("\n" + "=" * 80)
    print("[STEP 3] Searching for 'resistor' in EUR (European pricing)...")
    resistor_search = client.search_parts(
        query="resistor",
        limit=3,
        currency="EUR",
        country="GB"
    )

    resistor_results = resistor_search.get("data", {}).get("results", [])
    resistor_total = resistor_search.get("data", {}).get("total_hits", 0)
    
    print(f"✓ Found {resistor_total:,} resistors")
    print(f"✓ Displaying {len(resistor_results)} results with EUR pricing\n")
    
    for i, resistor in enumerate(resistor_results, 1):
        print(f"  [{i}] {resistor.get('name', 'Unknown')}")
        print(f"      MPN: {resistor.get('mpn', 'N/A')}")
        
        sellers = resistor.get("sellers", [])
        if sellers:
            for seller in sellers[:2]:  # Show 2 sellers
                seller_name = seller.get("name", "Unknown")
                offers = seller.get("offers", [])
                if offers:
                    offer = offers[0]
                    moq = offer.get("moq", 1)
                    inventory = offer.get("inventory_level", 0)
                    prices = offer.get("prices", [])
                    if prices:
                        price = prices[0].get("converted_price", 0)
                        currency = prices[0].get("converted_currency", "EUR")
                        print(f"      {seller_name}: {format_price(price, currency)} (MOQ: {moq}, Stock: {inventory})")
        print()

    # Step 4: Search for a specific MPN and get its details
    print("=" * 80)
    print("[STEP 4] Searching for specific MPN 'RC0805FR-071K1L'...")
    
    mpn_detail = client.get_part_detail(
        mpn="RC0805FR-071K1L",
        manufacturer="Yageo",
        currency="USD",
        country="US"
    )
    
    mpn_part = mpn_detail.get("data", {})
    if mpn_part.get("id"):
        print(f"✓ Found part by MPN")
        print(f"  Name: {mpn_part.get('mpn', 'N/A')}")
        print(f"  Manufacturer: {mpn_part.get('manufacturer', {}).get('name', 'N/A')}")
        print(f"  Slug: {mpn_part.get('slug', 'N/A')}")
        
        descriptions = mpn_part.get("descriptions", [])
        if descriptions:
            print(f"  Description: {descriptions[0].get('text', 'N/A')}")
        
        # Show best available resources
        best_image = mpn_part.get("best_image", {})
        if best_image.get("url"):
            print(f"  Image: {best_image.get('url')}")
        
        best_datasheet = mpn_part.get("best_datasheet", {})
        if best_datasheet.get("url"):
            print(f"  Datasheet: {best_datasheet.get('url')}")

    print("\n" + "=" * 80)
    print("WORKFLOW COMPLETE")
    print("=" * 80)


if __name__ == "__main__":
    main()