Product Data SDK

from datafiniti import DatafinitiProductSDK

sdk = DatafinitiProductSDK(api_key="API_KEY_HERE")

"""
Categories example: match products by category and inspect the `categories`
field on the Datafiniti Product API.

Every product carries a `categories` field — a list of the raw category labels
sellers and retailers have applied to it. You can search against this field to
pull products of a certain type. This script builds a small wine database by
matching `categories:wines`, then prints each product's brand, name, and the
full list of categories so you can see exactly how products get classified.

A note from the guide: category labels are messy real-world data. Searching
`categories:wine` (singular) also returns ice buckets and accessories, while
`categories:wines` (plural) does a much better job of returning actual wine
products. Inspecting the `categories` field, as this script does, is how you
discover which label works best for your use case.

See: https://docs.datafiniti.co/docs/build-a-database-of-wines
"""

from datafiniti.product import DatafinitiProductSDK
from datafiniti.errors import DatafinitiAPIError

# Build the SDK client from environment variables. `from_env()` reads your
# Datafiniti API token from the environment (DATAFINITI_API_TOKEN) so you don't
# have to hardcode credentials in the script.
sdk = DatafinitiProductSDK.from_env()

print("
=== PRODUCTS BY CATEGORY: WINES ===")

records_seen = 0

try:
    # paginate() yields one product record (a dict) at a time, fetching more
    # pages from the API as needed.
    #
    #   query       - "categories:wines" matches any product whose categories
    #                 field contains the label "wines". This is the core of
    #                 category-based matching. The "-categories:(...)" clause
    #                 uses the minus operator to exclude products labeled as
    #                 vinegars, oils, liqueur, gin, or tequila, which otherwise
    #                 slip in alongside wines.
    #   page_size   - records requested per API call.
    #   max_records - stop after 10 products, even if the query matches more.
    for product_record in sdk.paginate(
        query="categories:wines AND -categories:(vinegars OR oils OR liqueur OR gin OR tequila)",
        page_size=10,
        max_records=10,
    ):
        records_seen += 1

        # Use .get() with defaults so missing fields don't raise a KeyError.
        name = product_record.get("name", "Unknown Product")
        if len(name) > 75:
            name = name[:72] + "..."
        brand = product_record.get("brand", "Unknown Brand")

        # `categories` comes back as a list of label strings. Join them into a
        # single comma-separated line for display. Fall back to an empty list so
        # the join works even when the field is absent.
        categories = product_record.get("categories", [])
        category_list = ", ".join(categories) if categories else "None"

        print(f"
{records_seen}. {name}")
        print(f"   Brand:       {brand}")
        print(f"   Categories:  {category_list}")

    print(f"
Total Retrieved: {records_seen}")
except DatafinitiAPIError as e:
    # Raised when the API rejects the request (bad token, malformed query, rate
    # limit, etc.). `status_code` is the HTTP status; `errors` is a list of
    # human-readable messages explaining what went wrong.
    print(f"
API Error {e.status_code}: {e}")
    for msg in e.errors:
        print(f"  {msg}")

"""
Pagination example: pull pet food pricing data from the Datafiniti Product API.

The Datafiniti product database can match millions of records, far more than a
single API response returns. `paginate()` handles that for you by requesting one
page at a time and yielding records as it goes, so you can loop over a large
result set without managing offsets or page tokens yourself.
"""

from datafiniti.product import DatafinitiProductSDK
from datafiniti.errors import DatafinitiAPIError

# Build the SDK client from environment variables. `from_env()` reads your
# Datafiniti API token from the environment (DATAFINITI_API_TOKEN) so you don't
# have to hardcode credentials in the script.
sdk = DatafinitiProductSDK.from_env()

print("
=== PET FOOD PRICING PAGINATION TEST ===")

# Running tally of how many records we've printed across all pages.
records_seen = 0

try:
    # paginate() returns a generator. Each loop iteration gives you one product
    # record (a dict), automatically fetching the next page from the API behind
    # the scenes once the current page is exhausted.
    #
    #   query       - Datafiniti query syntax. This one matches products in the
    #                 "pet supplies" taxonomy, categorized as food, that have
    #                 pricing data ("prices:*" means the prices field exists).
    #   page_size   - how many records to request per API call (per page).
    #   max_records - safety cap on total records pulled, so the loop stops even
    #                 if the query matches far more. Useful while testing.
    for product_record in sdk.paginate(
        query='taxonomy:"pet supplies" AND categories:food AND prices:*',
        page_size=10,
        max_records=25,
    ):
        records_seen += 1
        # Each record is a dict of product fields. Use .get() so a missing field
        # falls back to a default instead of raising a KeyError.
        print(f"{records_seen}. {product_record.get('name', 'Unknown Product')}")
    print(f"
Total Retrieved: {records_seen}")
except DatafinitiAPIError as e:
    # Raised when the API rejects the request (bad token, malformed query, rate
    # limit, etc.). `status_code` is the HTTP status; `errors` is a list of
    # human-readable messages explaining what went wrong.
    print(f"
API Error {e.status_code}: {e}")
    for msg in e.errors:
        print(f"  {msg}")

"""
Pricing recommendations example: pull a list of products with their latest
price and availability from the Datafiniti Product API.

Each product record carries a set of "most recent price" fields that summarize
the latest pricing Datafiniti has seen for that product, so you don't have to
dig through the full historical `prices` array yourself. This script fetches up
to 10 electronics products that have pricing data and prints each one's
availability and most recent price.

See: https://docs.datafiniti.co/docs/provide-personalized-pricing-recommendations
"""

from datafiniti.product import DatafinitiProductSDK
from datafiniti.errors import DatafinitiAPIError

# Build the SDK client from environment variables. `from_env()` reads your
# Datafiniti API token from the environment (DATAFINITI_API_TOKEN) so you don't
# have to hardcode credentials in the script.
sdk = DatafinitiProductSDK.from_env()

print("
=== PRODUCT PRICING LIST ===")

records_seen = 0

try:
    # paginate() yields one product record (a dict) at a time, fetching more
    # pages from the API as needed.
    #
    #   query       - matches products categorized as electronics that have
    #                 pricing data ("prices:*" means the prices field exists).
    #   page_size   - records requested per API call.
    #   max_records - stop after 10 products, even if the query matches more.
    for product_record in sdk.paginate(
        query="categories:electronics AND prices:*",
        page_size=10,
        max_records=10,
    ):
        records_seen += 1

        # Pull the product name and the "most recent price" summary fields.
        # Use .get() with defaults so missing fields don't raise a KeyError.
        name = product_record.get("name", "Unknown Product")
        # Truncate long product names to 75 characters, ending with "..." so the
        # output stays readable.
        if len(name) > 75:
            name = name[:72] + "..."
        brand = product_record.get("brand", "Unknown Brand")
        availability = product_record.get(
            "mostRecentPriceAvailability", "Unknown"
        )
        amount = product_record.get("mostRecentPriceAmount")
        currency = product_record.get("mostRecentPriceCurrency", "USD")
        price_date = product_record.get("mostRecentPriceDate", "N/A")

        # Format the price only when we actually have an amount.
        if amount is not None:
            price = f"{amount} {currency}"
        else:
            price = "No price available"

        print(f"
{records_seen}. {name}")
        print(f"   Brand:         {brand}")
        print(f"   Availability:  {availability}")
        print(f"   Latest Price:  {price}")
        print(f"   Price Date:    {price_date}")

    print(f"
Total Retrieved: {records_seen}")
except DatafinitiAPIError as e:
    # Raised when the API rejects the request (bad token, malformed query, rate
    # limit, etc.). `status_code` is the HTTP status; `errors` is a list of
    # human-readable messages explaining what went wrong.
    print(f"
API Error {e.status_code}: {e}")
    for msg in e.errors:
        print(f"  {msg}")

"""
Product Views Example: Custom View On the Fly

A "view" lets you control exactly which fields the Datafiniti API returns for
each product, instead of getting the full product schema back. This keeps
responses small and focused on just the data you care about.

This script builds a custom view inline (no setup required) to fetch a handful
of fields for SteelSeries computer peripherals on Amazon that have reviews,
then prints each product's brand, identifiers, colors, and latest pricing.

References:
https://docs.datafiniti.co/docs/product_all
https://docs.datafiniti.co/docs/creating-a-custom-product-view-on-the-fly
"""

import json

from datafiniti.product import DatafinitiProductSDK
from datafiniti.errors import DatafinitiAPIError

# ---------------------------------------------------------------------------
# View options
#
# OPTION 1 (used below): Custom inline view
#   Pass a list of field objects directly in the request. No setup required.
#   - Simple fields are referenced by name only, e.g. {"name": "brand"}.
#   - Nested/complex fields (like the prices array) use "flatten": True plus a
#     "sub_fields" list naming the nested properties you want.
#
# OPTION 2: Premade (named) view — pass a string instead of a list.
#   "default"  All fields in the product schema (same as omitting view).
#   Example: sdk.search(query, view="default", num_records=10)
#   Reference: https://docs.datafiniti.co/docs/product_all
# ---------------------------------------------------------------------------

PRODUCT_VIEW = [
    {"name": "brand"},
    {"name": "name"},
    {"name": "manufacturerNumber"},
    {"name": "colors"},
    {"name": "mostRecentPriceAmount"},
    # Flatten the nested prices array and keep only a few sub-fields.
    {"name": "prices", "flatten": True, "sub_fields": [
        {"name": "amountMin"},
        {"name": "amountMax"},
        {"name": "currency"},
        {"name": "availability"},
        {"name": "isSale"},
    ]},
]

QUERY = 'domains:amazon AND categories:"Computer Peripherals" AND brand:"SteelSeries" AND reviews:*'

def format_list(value):
    """Render a list field as a comma-separated string, or 'N/A' if empty."""
    if isinstance(value, list) and value:
        return ", ".join(str(v) for v in value)
    return "N/A"

if __name__ == "__main__":
    # Build the SDK client from environment variables. `from_env()` reads your
    # Datafiniti API key from the environment (DATAFINITI_API_KEY).
    sdk = DatafinitiProductSDK.from_env()

    print("
=== PRODUCT CUSTOM VIEW ===
")

    # Show the exact request search() will send, so you can see how the query,
    # view, and other parameters are assembled into the API call. This mirrors
    # the payload built inside the SDK (see BaseDatafinitiSDK.search).
    request_url = f"{sdk.BASE_URL}/search"
    request_payload = {
        "query": QUERY,
        "num_records": 10,
        "format": "JSON",
        "view": PRODUCT_VIEW,
    }

    try:
        # search() returns one response containing up to num_records products.
        # Passing `view` limits the returned fields to those in PRODUCT_VIEW.
        response = sdk.search(QUERY, num_records=10, view=PRODUCT_VIEW)
        records = response.get("records", [])

        if not records:
            print("No products found.")

        for i, product in enumerate(records, start=1):
            name = product.get("name", "Unknown Product")
            if len(name) > 75:
                name = name[:72] + "..."

            print(f"{i}. {name}")
            print(f"   Brand:           {product.get('brand', 'N/A')}")
            print(f"   Mfr Number:      {product.get('manufacturerNumber', 'N/A')}")
            print(f"   Colors:          {format_list(product.get('colors'))}")
            print(f"   Latest Price:    {product.get('mostRecentPriceAmount', 'N/A')}")
            print()
    except DatafinitiAPIError as e:
        # Raised when the API rejects the request (bad key, malformed query,
        # rate limit, etc.). `status_code` is the HTTP status; `errors` is a
        # list of human-readable messages explaining what went wrong.
        print(f"
API Error {e.status_code}: {e}")
        for msg in e.errors:
            print(f"  {msg}")