Track Real SellerMagnet API Credit Spend: A SQLite Usage Dashboard

To track real SellerMagnet API credit spend, log the billed-or-free outcome of every attempt, not just the final result of a call, to a local SQLite table as it happens, then group by day and endpoint for a report. A retry helper that only returns the final result or raises the final error hides how many billed attempts happened along the way, so the ledger has to sit inside the retry loop itself, not around it.

Why can't get_with_retry answer how many credits a call cost?

get_with_retry from the Python quickstart exists to make retries transparent to the caller: it returns the data on success or raises ApiError on a final failure, and the backoff between attempts happens inside the loop. That is exactly why it cannot answer a billing question by itself - the caller never sees the 502 that got retried and billed before the third attempt finally succeeded. Logging has to happen on every attempt, inside the same loop, before the loop decides whether to retry.

How do I build a credit-usage ledger in Python?

tracked_call() runs the same policy as retry.py's get_with_retry (same RETRYABLE set, same backoff), but calls _classify() on every attempt's status, message and Retry-After before deciding whether to retry, and writes the outcome to a calls table keyed by day and endpoint. get and ApiError are the unmodified functions from sellermagnet.py; nothing about the actual HTTP call is reimplemented, only the logging that wraps it.

usage_tracker.py - classify and log every attempt, then report per day

"""Log every SellerMagnet API attempt to a local SQLite ledger and print a per-day cost report.

`get_with_retry` (retry.py) hides how many billed attempts happened inside its own backoff loop: a
caller that only checks the final return value or the final exception has no idea a 502 was retried
twice (2 credits) before it finally raised. This module runs the SAME retry policy (same RETRYABLE
set, same backoff) but logs the billing outcome of every attempt BEFORE deciding whether to retry, so
the ledger matches what the account was actually charged, not what the caller happened to see.

Classification rules (SellerMagnet API's documented status codes and billing behaviour):
  200                      -> billed (1 credit)
  404                      -> billed (a real "not found" answer, not an outage)
  500 WITHOUT Retry-After  -> refunded (free); WITH Retry-After it stays charged
  502 / 503                -> billed on every attempt
  401 / 403 / 429          -> free (rejected before the credit deduction)
  400                      -> billed ONLY for the known endpoint-specific cases raised after the
                              deduction (count over 50, a non-integer count/offset, a malformed ASIN
                              on statistics/estimated-sells, a bad converter input); every other 400
                              is a free validation gate (e.g. an unsupported marketplace, or an
                              unknown category ID, which moved from billed to free on 2026-11-01)
  anything else            -> assumed billed; an unrecognised status is new API behaviour, and a
                              ledger that silently calls it free would under-count real spend
"""
import sqlite3
import time
import random

from sellermagnet import ApiError, get
from retry import RETRYABLE

# endpoint -> substrings of the billed 400's message (these 400s are raised inside the view, AFTER
# the credit is deducted, so they are charged unlike most 400s). "Invalid CategoryId" on
# amazon-bestsellers is deliberately NOT here: since commit c2e6bf3 (2026-11-01) that check runs
# BEFORE the credit deduction, so it is a free 400 like any other gate - listing it here would make
# the ledger OVER-count, the opposite failure from the one this tool exists to avoid.
BILLED_400_HINTS = {
    "amazon-search": ("Count Max", "must be a whole number"),
    "amazon-bestsellers": ("Count Max", "must be a whole number", "offset + count must be at most"),
    "amazon-deals": ("Count Max", "must be a whole number"),
    "amazon-product-statistics": "Invalid product ASIN",
    "amazon-product-search-estimated-sells": "Invalid product ASIN",
    "amazon-asin-converter": "Invalid",
}


def _classify(endpoint: str, status: int, message: str, retry_after: float) -> tuple[bool, str]:
    """Return (billed, note) for one HTTP attempt. Never guesses from memory: every branch cites the
    documented billing rule it encodes, so a future billing change is one line to fix, not a rewrite."""
    if status == 200:
        return True, "success"
    if status == 404:
        return True, "not found (real answer)"
    if status in (502, 503):
        return True, "upstream error, billed per attempt"
    if status == 500:
        return (True, "500 kept Retry-After, stays charged") if retry_after else (False, "refunded")
    if status in (401, 403, 429):
        return False, "free gate"
    if status == 400:
        hints = BILLED_400_HINTS.get(endpoint, ())
        hints = (hints,) if isinstance(hints, str) else hints
        if any(h.lower() in (message or "").lower() for h in hints):
            return True, f"billed 400 ({endpoint})"
        return False, "free gate 400"
    return True, f"unrecognised status {status}, assumed billed"


def init_db(path: str) -> sqlite3.Connection:
    db = sqlite3.connect(path)
    db.execute(
        "CREATE TABLE IF NOT EXISTS calls ("
        "id INTEGER PRIMARY KEY, day TEXT, endpoint TEXT, status INTEGER, billed INTEGER, note TEXT)"
    )
    db.commit()
    return db


def _log(db: sqlite3.Connection, day: str, endpoint: str, status: int, billed: bool, note: str) -> None:
    db.execute(
        "INSERT INTO calls (day, endpoint, status, billed, note) VALUES (?,?,?,?,?)",
        (day, endpoint, status, int(billed), note),
    )
    db.commit()


def tracked_call(session, db: sqlite3.Connection, endpoint: str, day: str, attempts: int = 4, **params):
    """Same policy as retry.py's get_with_retry, logging every attempt's billing outcome first."""
    internal_errors = 0
    for attempt in range(1, attempts + 1):
        try:
            data = get(session, endpoint, **params)
            _log(db, day, endpoint, 200, True, "success")
            return data
        except ApiError as err:
            message = str(err).partition(": ")[2]
            billed, note = _classify(endpoint, err.status, message, err.retry_after)
            _log(db, day, endpoint, err.status, billed, note)
            internal_errors += err.status == 500
            if err.status not in RETRYABLE or attempt == attempts or internal_errors > 1:
                raise
            backoff = min(2 ** attempt, 30) + random.uniform(0, 1)
            time.sleep(max(backoff, err.retry_after))


def report(db: sqlite3.Connection, day: str) -> str:
    """One ASCII table: billed / free attempts and total credits spent per endpoint, for one day."""
    rows = db.execute(
        "SELECT endpoint, SUM(billed), COUNT(*) - SUM(billed), COUNT(*) FROM calls "
        "WHERE day = ? GROUP BY endpoint ORDER BY endpoint",
        (day,),
    ).fetchall()
    if not rows:
        return f"No calls logged for {day}."
    lines = [f"{'ENDPOINT':<40}{'BILLED':>8}{'FREE':>8}{'ATTEMPTS':>10}"]
    total = 0
    for endpoint, billed, free, attempts in rows:
        billed = billed or 0
        total += billed
        lines.append(f"{endpoint:<40}{billed:>8}{free:>8}{attempts:>10}")
    lines.append("-" * 66)
    lines.append(f"{'TOTAL CREDITS SPENT ON ' + day:<56}{total:>10}")
    return "\n".join(lines)

Illustrative run output: 3 lookups, a search call billed for exceeding count=50 then retried by the caller with a valid count, an offers 404, a refunded-then-retried statistics call

ENDPOINT                                  BILLED    FREE  ATTEMPTS
amazon-product-lookup                          3       0         3
amazon-product-offers                          1       0         1
amazon-product-statistics                      1       1         2
amazon-search                                  2       0         2
------------------------------------------------------------------
TOTAL CREDITS SPENT ON 2026-11-11                                7

How is this different from the API cost calculator?

The pricing calculator estimates cost BEFORE calling anything: it answers "what would 10,000 lookups a month cost on this plan". This ledger answers a different question AFTER the fact: "what did this account actually spend today, and on which endpoint", including the retries a pre-call estimate has no way to predict because it depends on how often upstream actually returns a 502 or 503 on a given day.

What does the daily report look like?

report() groups the day's rows by endpoint and sums the billed column, so the last line is always the total credits actually spent that day. Nothing here calls the API: the report reads only the local SQLite table that tracked_call() already wrote to, so running it costs zero credits and works offline.

The query behind report() (SQLite)

SELECT endpoint, SUM(billed), COUNT(*) - SUM(billed), COUNT(*)
FROM calls WHERE day = ?
GROUP BY endpoint ORDER BY endpoint;

Bottom line: log the billed-or-free outcome of every attempt inside the retry loop, not around it, classify a 400 by endpoint and message rather than trusting the status code alone, and default an unrecognised status to billed so the ledger never quietly under-counts real spend. For planning a budget before making a single call, use the pricing calculator; a free account includes 150 credits to try both.