isbnator

Async Python client for the ISBNdb REST API v2 with built-in caching (file / DynamoDB / none) and rate limiting.

Installation

pip install -r requirements.txt

Dependencies: httpx, pydantic, dynamorator, logorator, python-dotenv

API Key

Resolved in order:

1. api_key constructor argument
2. ISBNDB_TOKEN environment variable
3. Raises ISBNdbConfigError if neither is set

Client Initialization

from isbnator import ISBNdbClient

client = ISBNdbClient(
    api_key="<key>",               # or set ISBNDB_TOKEN env var
    cache="file",                  # "file" | "dynamodb" | "none" (default: "none")
    cache_dir=".isbndb_cache",     # directory for file cache (default: ".isbndb_cache")
    dynamo_table="isbndb-cache",   # DynamoDB table name (required if cache="dynamodb")
    cache_ttl_days=90,             # cache TTL in days (default: 90)
    max_concurrent=5,              # max parallel requests in batch (default: 5)
    max_retries=3,                 # max retries per request (default: 3)
    requests_per_second=1,         # per-second throttle (default: 1)
    request_timeout=30,            # request timeout in seconds (default: 30)
    rate_limit_warn_threshold=100, # warn when remaining quota drops below this (default: 100)
)

# Use as async context manager
async with ISBNdbClient(cache="file") as client:
    ...

Models

SearchQuery (input for batch_search)

class SearchQuery(BaseModel):
    title: str                   # required — search text
    author: str | None = None    # optional — filter by author

Book (parsed from ISBNdb responses)

class Book(BaseModel):
    title: str | None = None
    title_long: str | None = None
    isbn: str | None = None           # ISBN-10 or ISBN-13 (ISBNdb is inconsistent)
    isbn13: str | None = None         # ISBN-13
    authors: list[str] = []
    publisher: str | None = None
    date_published: str | None = None # format varies: "2015-05-12", "2015", "May 2015"
    binding: str | None = None        # e.g. "Paperback", "Hardcover", "Audio CD"
    pages: int | None = None
    dimensions: str | None = None
    image: str | None = None          # cover image URL
    language: str | None = None       # e.g. "en", "es"
    edition: str | None = None
    subjects: list[str] = []
    synopsis: str | None = None
    msrp: float | None = None

All fields are optional. Unknown fields from ISBNdb are silently ignored (extra="ignore").

QueryResult (returned by all query methods)

class QueryResult(BaseModel):
    status: str              # "success" | "not_found" | "error"
    books: list[Book] = []   # parsed Book models
    from_cache: bool = False # True if result was served from cache
    error: str | None = None # error message when status == "error"
    data: dict | None = None # raw JSON response from ISBNdb (None when from cache)

RateLimit (returned by quota())

class RateLimit(BaseModel):
    daily_limit: int = 5000
    used: int = 0
    remaining: int = 5000

Methods

All methods are async and return QueryResult unless noted.

book(isbn: str) -> QueryResult

Lookup a single book by ISBN-10 or ISBN-13.

result = await client.book("9780143127550")
# result.status == "success"
# result.books == [Book(title="Everything I Never Told You - A Novel", ...)]

books(query, column=None, page=1, page_size=20) -> QueryResult

Full-text search. column can be "title", "author", "date_published", etc.

result = await client.books("Gatsby", column="title")
# result.books == [Book(...), Book(...), ...]  (up to page_size)

author(name: str) -> QueryResult

Search by author name.

result = await client.author("F. Scott Fitzgerald")

publisher(name: str) -> QueryResult

Search by publisher name.

result = await client.publisher("Penguin")

search(**kwargs) -> QueryResult

Multi-field search via /search/books. Supported kwargs: text, author, isbn, isbn13, subject, publisher. None values are excluded.

result = await client.search(text="Great Gatsby", author="Fitzgerald")

batch_search(queries, on_progress=None) -> list[QueryResult]

Primary use case. Sends multiple queries with controlled concurrency. Returns results in the same order as input.

from isbnator import SearchQuery

queries = [
    SearchQuery(title="The Great Gatsby", author="F. Scott Fitzgerald"),
    SearchQuery(title="1984", author="George Orwell"),
    SearchQuery(title="xyznonexistent"),
]

results = await client.batch_search(queries)
# len(results) == len(queries), same order

for query, result in zip(queries, results):
    print(query.title, result.status, len(result.books), result.from_cache)
    # "The Great Gatsby" "success" 20 False
    # "1984"             "success" 20 False
    # "xyznonexistent"   "not_found" 0 False

Behavior:

• Checks cache first (batch), only queries ISBNdb for misses
• Runs misses with max_concurrent parallelism, throttled by requests_per_second
• Caches both successful and not_found results (negative caching)
• Each query is cached immediately when its API response arrives (not after the full batch)
• Failed queries return QueryResult(status="error", error="...") — never raises

Progress callback

Pass an optional on_progress callback to receive events as the batch progresses. The callback is synchronous (not awaited) and receives a plain dict.

results = await client.batch_search(queries, on_progress=lambda e: print(e))

Callback signature: Callable[[dict], None] | None

Events emitted:

batch_cache_checked — emitted once after the initial cache lookup:

{
    "event": "batch_cache_checked",
    "total": 11,       # total queries in the batch
    "cached": 4,       # how many were found in cache
    "to_fetch": 7,     # how many need API calls
}

query_complete — emitted after each individual query finishes (cache hit or API call):

{
    "event": "query_complete",
    "completed": 3,          # how many queries done so far
    "total": 11,             # total queries in the batch
    "query": SearchQuery,    # the SearchQuery that was processed
    "status": "success",     # "success" | "not_found" | "error"
    "from_cache": False,     # True if result came from cache
    "books": 20,             # number of books in the result
}

Example with SSE streaming:

async with ISBNdbClient(cache="dynamodb", dynamo_table="my-table") as client:
    results = await client.batch_search(
        queries,
        on_progress=lambda e: emitter.emit(e),
    )

quota() -> RateLimit

Fetches API usage stats from /stats.

q = await client.quota()
# q.daily_limit == 5000, q.used == 42, q.remaining == 4958

Also accessible as client.rate_limit property (updated after calling quota()).

close()

Closes the underlying HTTP client. Called automatically when using async with.

Caching

Three cache backends, selected via cache constructor param.

Backend	cache=	Notes
None	"none"	Every call hits ISBNdb (default)
File	"file"	One JSON file per query in cache_dir
DynamoDB	"dynamodb"	Via dynamorator.DynamoDBStore, requires dynamo_table

Cache keys are SHA-256 hashes of the endpoint path + sorted query params. Both successful and not_found results are cached with the configured TTL. File cache checks expiry on read and deletes stale entries.

Exceptions

ISBNdbError              # base exception
├── ISBNdbConfigError    # missing API key or bad config
├── ISBNdbRateLimitError # daily quota exhausted (remaining == 0)
└── ISBNdbAPIError       # non-retryable API error
    .status_code: int
    .message: str

Exceptions are raised for config issues and quota exhaustion. In batch_search, individual query failures are captured in QueryResult(status="error"), not raised.

Retry Logic

Retries on: HTTP 429, 5xx, connection errors, timeouts.

• Exponential backoff with jitter (1s, 2s, 4s base)
• Respects Retry-After header on 429
• After max_retries exhausted: raises ISBNdbAPIError (single queries) or returns error QueryResult (batch)

Imports

from isbnator import (
    ISBNdbClient,
    SearchQuery,
    QueryResult,
    Book,
    BooksResponse,
    RateLimit,
    ISBNdbError,
    ISBNdbConfigError,
    ISBNdbRateLimitError,
    ISBNdbAPIError,
)