keble-keepa 1.2.2

No project description provided

Keble-Keepa

A powerful Python wrapper for the Keepa API with enhanced data analysis capabilities and caching.

Version 1.2.2 update

• Enriched keble_keepa.testing with keepa_raw_fixture(...), a package-data loader for checked-in Keepa payload fixtures.
• Added reusable contract fixture shapes for:
  • normal product
  • no buy box
  • missing title
  • variation parent
  • variation child
  • suppressed listing
  • unstable sales rank
  • low-data product
  • high-data product
• Downstream repos should add missing reusable Keepa shapes here first instead of creating one-off raw payload mocks in business packages.

Version 1.2.1 update

• Release metadata fix: removed the legacy custom package classifier rejected by PyPI upload validation.
• 1.2.1 contains the same testing-toolkit behavior introduced in 1.2.0; the version bump keeps the pushed v1.2.0 tag immutable after the publish blocker was discovered.

Version 1.2.0 update

• Added keble_keepa.testing as the shared Keepa test toolkit for downstream repos:
  • FakeKeepaGateway
  • keepa_product_snapshot(...)
  • keepa_raw_payload(...)
  • keepa_raw_fixture(...)
  • keepa_variation_snapshot(...)
  • Keepa cassette secret filters
  • requires_keepa_live
  • Keepa payload/schema assertion helpers
• Registered the cross-repo Keble pytest marker vocabulary in pyproject.toml.
• Default tests no longer require .env, Mongo, Redis, or a Keepa API token. Mongo-backed cache tests skip when MONGO_DB_URI is absent, and live Keepa tests skip unless RUN_KEEPA_LIVE=1.
• Replaced session-wide pytest cache deletion with per-test Keepa cache collection isolation and teardown.
• Added pyrightconfig.json and verified npx --yes pyright ..
• Side effect if changes:
  • Downstream repos should use keble_keepa.testing instead of ad-hoc raw Keepa mocks.
  • Default development tests are safe to run without provider keys or local DB services.
  • Live Keepa drift checks remain opt-in through markers and env guards.

Testing

Default fast test command:

uv run pytest -m "not live and not slow and not eval and not local_stack and not db_stack and not container"

Run the offline testing toolkit tests:

uv run pytest tests/test_testing

Run Mongo-backed cache integration tests:

MONGO_DB_URI='mongodb://localhost:27017' uv run pytest -m "integration and mongo"

Run live Keepa tests:

RUN_KEEPA_LIVE=1 KEEPA_API_TOKEN='<token>' uv run pytest -m keepa_live

Run static syntax/type checking:

npx --yes pyright .

Use FakeKeepaGateway and product factories for downstream unit tests:

from keble_keepa import DomainId, RequestProductsQuery
from keble_keepa.testing import FakeKeepaGateway, keepa_product_snapshot, keepa_raw_fixture


gateway = FakeKeepaGateway(
    products=[
        keepa_product_snapshot(
            asin="B08TEST001",
            title="Example Product",
            monthly_sold=1200,
            buy_box_price=2399,
        )
    ]
)

response = gateway.request_products(
    RequestProductsQuery(asin=["B08TEST001"], domain=DomainId.COM)
)
assert response.products[0].monthly_sold == 1200

raw_contract_payload = keepa_raw_fixture("product_variation_parent")
assert raw_contract_payload["variationCSV"] == "B000CHILD01,B000CHILD02"

Live Keepa calls must never be added to the default suite. Mark them with @pytest.mark.live, @pytest.mark.keepa_live, and requires_keepa_live.

Version 1.1.16 update

• Async Keepa calls now create aiohttp sessions with trust_env=True, so local Docker backends and workers honor standard HTTP_PROXY / HTTPS_PROXY environment variables without hard-coding proxy values into package or production config.
• The sync requests paths are unchanged because requests already honors proxy environment variables by default.
• Side effect if changes:
  • Backend positioning fresh-items and product-discovery/report-generation flows use async Keepa calls through this package.
  • Local live E2E coverage depends on Docker proxy env being respected while keeping ALL_PROXY unset.

Version 1.1.12 update

• Cache-poisoning fix: per-key cache reads (get_from_cache, aget_from_cache, aget_request_products_cache_parts) now return the NEWEST row (created DESC, _id DESC tiebreak). Previously the unsorted read returned the oldest row, so a negative data=None row written from one partial Keepa response shadowed every later good fetch for the whole cache window.
• New skip_cache_read=True on KeepaApi.arequest_products / _aget: forces a live Keepa fetch (cache read bypassed) while still writing the fresh response to cache, so a forced fetch repairs a poisoned key for every later caller.
• ready_mongo now also creates a {"key": 1, "created": -1} index — the per-key read previously had no supporting index at all.
• Negative (data=None) rows are still written and still count as known negative hits; only their precedence vs newer rows changed.

Version 1.1.11 update

• Pricing bugfix: Product.average_price_int no longer falls back to the all-time arithmetic mean when the NEW price has been stable for longer than the recent 3-month window (production case: a stable $19.95 product reported as $97.95 because only years-old high-price change points existed).
• New UnitDataList.time_weighted_mean(gte=..., lte=...): duration-weighted average for Keepa change-point series. It forward-fills the last change point recorded before the window start and weights every value by how long it stayed effective, so periods with frequent price changes no longer dominate the mean. UnitDataList.mean (plain arithmetic mean) is unchanged.
• Known limitation (unchanged from previous versions): negative Keepa values (-1, offer unavailable) are dropped at UnitDataList.build, so out-of-stock gaps still count as the previous price's effective duration.

Version 1.1.10 update

• Release-hygiene follow-up to 1.1.9 (no public API change):
  • Product._is_valid_keepa_timestamp is now a TypeGuard[int | float], so _extract_first_keepa_timestamp_from_series narrows the validated value without a duplicate isinstance check (clears the last pyright error).
  • tests/conftest.py now wipes the Keepa pytest Mongo cache collection at pytest_sessionfinish too, symmetric with pytest_sessionstart.
  • Regenerated uv.lock to track the bumped package version.

Version 1.1.9 update

• Type-safety fixes in the prompt-snapshot path:
  • KeepaApiBatchLoader._retry_exception guards tenacity's optional retry_state.outcome explicitly (no getattr / try-except) so retry logging is type-clean.
  • _dimension_triple_cm narrows item/package dimension tuples to tuple[float, float, float] | None, used by KeepaProductPromptPhysicalFacts.from_product.
• Rating rendering: _prompt_metric_value_text always converts the Keepa 0..50 rating unit to a 0.0..5.0 star score (value / 10), fixing the rare sub-0.5-star case (Keepa 5 now renders 0.5/5, not 5.0/5).
• Robustness: _prompt_model_value raises KeyError for any name that is not a declared model field, turning a field typo / removed field into a loud failure instead of a value that silently nulls forever. Declared-but-absent fields on partial model_construct fixtures still return None.
• Coverage: added a full-range rating-scale parametrize, an unknown-field guard test, and a real-world tests/test_api/test_objects/test_product_prompt_snapshot_irl.py that asserts the prompt snapshot populates physical facts, latest metrics, sales ranks, rating, and reviews from live Keepa products (guards upstream Keepa field drift that the field-name check cannot detect).

Version 1.1.6 update

• Added KeepaProductPromptSnapshot, bounded prompt series/point/rank schemas, and Product.to_prompt_snapshot(...) / Product.to_prompt_markdown(...).
• Raw Keepa Product remains the storage/API truth; prompt snapshots are the canonical LLM/RAG surface and intentionally exclude raw csv, salesRanks, monthly-sold histories, offers, review payloads, and unbounded variations.
• Snapshot history defaults to a 7-day recent window with explicit kwargs for recent-window days, per-series caps, feature caps, variation caps, category path caps, image inclusion, and deterministic now.
• Added regression coverage for latest product facts, window/cap overrides, raw history exclusion, and B07R295MLS-like prompt-budget behavior.

Version 1.1.5 update

• Product.tracking_since_datetime now exposes Keepa's trackingSince as a deterministic release-date fallback when official listing dates are unavailable.
• Product.guessed_has_offer_since_datetime now uses valid trackingSince before scanning historical series, which keeps new ASINs with listedSince=0 from showing an empty release date.
• Added regression coverage for structured-image-compatible ASINs whose official release fields are missing.

Version 1.1.4 update

• KeepaApiBatchLoader.max_concurrent_tasks is now enforced for product batch requests with an async semaphore.
• max_concurrent_tasks is validated as a positive integer, matching the existing product batch-size validation.
• Added regression coverage proving larger ASIN loads do not exceed the configured product request concurrency.

Version 1.1.3 update

• Added explicit sync and async HTTP timeout configuration to KeepaApi.
• Product batch loading now defaults to smaller 20-ASIN Keepa requests and keeps the existing split-and-continue fallback for exhausted retries.
• Partial product cache lookup now reuses cached ASIN rows for exact request params and only sends missing ASINs back to Keepa.
• Existing product cache rows with data=None are treated as known negative cache hits, so they are not requested again for the same params.
• Added timeout, partial-cache, batch-size, split-fallback, and warning-log regression tests.

Version 1.1.2 update

• Added typed support for Keepa's structured images product field, which replaces deprecated imagesCSV.
• Product.image_filename, Product.image_filenames, Product.image_url, and Product.image_urls now prefer structured image filenames and fall back to legacy imagesCSV.
• Added regression tests covering structured large images, medium-image fallback, legacy CSV fallback, and URL generation.

Version 1.1.1 update

• Product.new_price_is_map now accepts missing newPriceIsMAP fields from Keepa as None.
• Explicit newPriceIsMAP=true and newPriceIsMAP=false payloads are still preserved as booleans.
• Added regression tests for missing and explicit MAP-restriction values.

Version 1.0.4 update

• Product.has_offer_since_datetime now ignores listedSince values that resolve to a future datetime.
• Product.guessed_has_offer_since_datetime now only uses non-future Keepa time-series timestamps.
• Added regression tests for future listedSince and all-future series edge cases.

Version 1.0.3 update

• Hardened KeepaApiBatchLoader.aload_asins_from_keepa against full-batch failure.
• When one Keepa ASIN batch fails after retries, loader now splits the batch into smaller chunks and continues best-effort loading.
• Single-ASIN hard failures are skipped with warnings instead of aborting the whole upstream task.
• Added regression tests for split-and-continue and single-ASIN skip behavior.

Version 1.0.2 update

• Added parameterized pytest coverage for Product.guessed_has_offer_since_datetime.
• Coverage now includes multiple CSV timestamp layouts, mixed keepaTime series, and invalid-series rejection cases.

Version 1.0.1 update

• Enhanced Product.guessed_has_offer_since_datetime to scan all Keepa-time series lists for the earliest valid timestamp.
• Added safeguards to ignore non-time-series list fields (for example category id lists).
• Added unit tests covering listed-since priority, multi-series earliest selection, and non-series filtering.

Overview

Keble-Keepa provides a convenient interface to interact with Keepa's API for accessing Amazon product data. The package includes:

• Complete API client with sync and async methods
• MongoDB-based caching system for efficient API usage
• Enhanced data models with additional properties and analysis methods
• Batch loading system for processing multiple products

Installation

pip install keble-keepa

Quick Start

from keble_keepa import KeepaApi, RequestProductsQuery, DomainId

# Initialize the API client
keepa_api = KeepaApi(
    mongo=mongo_client,
    api_token="your-keepa-api-token",
    mongo_database="keepa_cache_db",
    keepa_cache_collection="keepa_cache",
    cache_days=7,  # Cache data for 7 days
    request_timeout_secs=90,
    connect_timeout_secs=10,
    sock_read_timeout_secs=60,
)

# Request product data
products_data = keepa_api.request_products(
    RequestProductsQuery(
        asin=["B07B7K7N3P"],  # ASIN of the product
        domain=DomainId.COM,   # Amazon US
        stats=90,              # Get stats for the last 90 days
        offers=20,             # Get up to 20 marketplace offers
        rating=1,              # Include rating history
    )
)

# Access the product data
product = products_data.products[0]
print(f"Title: {product.title}")
print(f"Average price: ${product.average_price_int/100:.2f}")
print(f"Current rating: {product.latest_ratings_int/10:.1f}/5.0 ({product.latest_reviews} reviews)")

# Get size and weight information
size, weight = product.size_and_weight
if size:
    print(f"Size (L×W×H): {size[0]}×{size[1]}×{size[2]} cm")
if weight:
    print(f"Weight: {weight} g ({weight/453.59237:.2f} lb)")

KeepaApi

The main class for interacting with the Keepa API.

Initialization

from keble_keepa import KeepaApi
from pymongo import MongoClient

# Initialize
keepa_api = KeepaApi(
    mongo=MongoClient("mongodb://localhost:27017"),
    api_token="your-keepa-api-token",
    mongo_database="keepa_cache_db",
    keepa_cache_collection="keepa_cache",
    cache_days=7,
    request_timeout_secs=90,
    connect_timeout_secs=10,
    sock_read_timeout_secs=60,
)

Key Methods

# Sync Methods
keepa_api.request_products(query)             # Get product data by ASIN
keepa_api.product_finder(query)               # Search for products
keepa_api.product_search(query)               # Search for products by keywords
keepa_api.category_lookup(query)              # Look up category details
keepa_api.category_search(query)              # Search for categories
keepa_api.browsing_deal(query)                # Find deals on Amazon
keepa_api.lightning_deal(query)               # Find lightning deals
keepa_api.request_best_sellers(query)         # Get best seller lists
keepa_api.request_seller_information(query)   # Get seller information
keepa_api.retrieve_token_status()             # Check API token status

# Async Methods
await keepa_api.arequest_products(query, amongo=amongo, extended_aredis=aredis)
await keepa_api.aproduct_finder(query, amongo=amongo, extended_aredis=aredis)
await keepa_api.acategory_lookup(query, amongo=amongo, extended_aredis=aredis)
await keepa_api.acategory_search(query, amongo=amongo, extended_aredis=aredis)
# ... and more

Batch Loader

For loading multiple products efficiently:

from keble_keepa import KeepaApiBatchLoader

# Initialize
batch_loader = KeepaApiBatchLoader(
    keepa_api=keepa_api,
    max_concurrent_tasks=20,
    retry_sleep_secs=30,
    max_retry_per_request=3,
    product_batch_size=20,
)

# Load multiple products by ASINs
products = await batch_loader.aload_asins_from_keepa(
    amongo=amongo,
    extended_aredis=aredis,
    asins=["B07B7K7N3P", "B0113UZJE2"],
    marketplace=AmazonMarketplace.US,
    with_reviews=True,
)

# Load similar products based on a reference product
similar_products = await batch_loader.aload_similar_products(
    product=reference_product,
    category_id=category_id,
    amongo=amongo,
    extended_aredis=aredis,
    marketplace=AmazonMarketplace.US,
    search_same_category_asins_page_size=100,
    search_same_category_asins_max_page=10,
    search_same_category_tolerant_ratio=0.5,
    keyword="digital camera",
    with_reviews=True,
    # OPT-IN: widen the Keepa title search with singular/plural variants
    # (e.g. "berry" also searches "berries"). Default False keeps callers
    # unchanged; the leaf dedups and caps expanded terms at Keepa's 50-keyword
    # limit. Helpers: keble_keepa.singular_plural_variants / expand_singular_plural.
    expand_keyword_variants=True,
)

# Load similar products with a limit
limited_similar = await batch_loader.aload_similar_products_with_limit(
    product=reference_product,
    amongo=amongo,
    extended_aredis=aredis,
    marketplace=AmazonMarketplace.US,
    limit=50,
)

Request Schemas

RequestProductsQuery

For retrieving detailed product data by ASIN.

from keble_keepa import RequestProductsQuery, DomainId

query = RequestProductsQuery(
    domain=DomainId.COM,            # Amazon marketplace (1: US, 2: UK, 3: DE, etc.)
    asin=["B07B7K7N3P"],           # List of ASINs to fetch (up to 100)
    # Alternative: Use product codes
    # product_code=["885909950805"], # UPC, EAN, or ISBN-13 codes
    
    # Optional parameters (same as Keepa official doc)
    stats=90,                       # Get stats for the last 90 days
    update=1,                       # Force refresh if data is older than 1 hour
    offers=20,                      # Get up to 20 marketplace offers
    buybox=1,                       # Include buy box data
    rating=1,                       # Include rating history
    # ...and more
)

Other Query Types

• ProductFinderQuery: Advanced product search with filtering
• ProductSearchQuery: Simple keyword search
• CategoryLookupQuery: Category information lookup
• CategorySearchQuery: Search for categories
• RequestBestSellersQuery: Get best seller products
• RequestSellerInformationQuery: Get seller information

Response Schemas

Product Model

The core model representing Amazon product data with enhanced properties:

# Basic Keepa data fields (same as Keepa official doc)
product.asin                      # Amazon Standard Identification Number
product.title                     # Product title
product.domain_id                 # Amazon locale ID
product.manufacturer              # Manufacturer name
product.brand                     # Brand name
product.product_group             # Product category group
product.csv                       # CSV data with price/rank history
# ...many more fields

# Enhanced properties (not in Keepa API)
product.size_and_weight           # Tuple of (dimensions tuple, weight in grams)
product.average_price_int         # Average price as integer (100 = $1.00)
product.latest_ratings_int        # Current rating (0-50, integer)
product.latest_reviews            # Current number of reviews
product.marketplace               # Marketplace enum (US, UK, DE, etc.)
product.amazon_marketplace        # Amazon marketplace enum
product.currency                  # Currency enum
product.currency_symbol           # Currency symbol
product.monthly_sales_unit_data_list  # Monthly sales data
product.amazon_prices_unit_data_list  # Amazon price history
product.new_prices_unit_data_list     # 3rd party new price history
product.used_prices_unit_data_list    # Used price history
product.bsr_list                  # List of Best Seller Ranks
product.categories_ids            # List of category IDs
product.image_urls                # List of image URLs
product.has_offer_since_datetime  # First date product was available
product.is_bsr_stable             # Whether BSR is stable over time
product.baseline_monthly_sales    # Estimated minimum monthly sales
product.last_seen                 # Most recent datetime any data was updated
# ...and more

Other Response Types

• ProductFinderResponse: Results from product finder search
• ProductSearchResponse: Results from keyword search
• CategoryLookupResponse: Category details
• RequestBestSellersResponse: Best seller products

License

MIT

1.1.7 Usage Accounting Boundary

Async Keepa product finder and product request calls accept an optional UsageAccountingRecorderProtocol. Cache hits do not emit usage; live Keepa calls emit KEEPA unit-count events from the provider boundary that actually spent API capacity.

Batch loading uses tenacity for retry ownership and forwards the recorder into the canonical Keepa API methods, so higher-level packages do not duplicate Keepa accounting.

1.1.8 Retry Predicate Hardening

Keepa batch loading still uses tenacity, but retries are limited to transport and timeout failures from the HTTP stack. Validation, typing, and programming errors are allowed to fail immediately so tests and callers do not hide bad inputs behind retry loops.

Version 1.1.13 Update

Product Finder page sizing is now schema-owned. Callers may request a small result limit, but ProductFinderQuery.resolve_valid_finder_page(...) converts that request into Keepa-valid perPage/page values before the API call and the loader slices locally. This prevents chat/query tools from sending invalid finder windows while preserving caller-level limits.