BGG API client: disk cache, rate limiting, 202-queue retry, XML parsing
httpx client with injectable clock/sleep/rng for testability. Successful responses cached under data/bgg_cache/ keyed by endpoint+params; 202 retries follow the spec schedule (2/5/10/30s, give up after 5); 429/503 get jittered exponential backoff; consecutive requests are spaced rate_limit_seconds apart. Parsers (via defusedxml, per security hook) cover search, thing (+stats/+versions), and collection, including the Not Ranked and error-document cases. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Eric Wagoner
co-authored by
Claude Fable 5
Claude Fable 5
parent
d58568cceb
commit
4e1211feb6
@@ -0,0 +1,145 @@
|
||||
"""BGG XML API2 client: disk cache, rate limiting, 202-queue and 429/503 retry.
|
||||
|
||||
Every successful response is cached under data/bgg_cache/ keyed by
|
||||
endpoint + params, so re-runs never re-hit the API. The rate limiter
|
||||
guarantees ≤1 request every rate_limit_seconds to any BGG endpoint.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import random
|
||||
import re
|
||||
import time
|
||||
from collections.abc import Callable, Iterable
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlencode
|
||||
|
||||
import httpx
|
||||
|
||||
from bggpipe.models import (
|
||||
CollectionItem,
|
||||
SearchResult,
|
||||
ThingDetails,
|
||||
parse_collection,
|
||||
parse_search,
|
||||
parse_things,
|
||||
)
|
||||
|
||||
BASE_URL = "https://boardgamegeek.com/xmlapi2"
|
||||
QUEUE_BACKOFF = (2.0, 5.0, 10.0, 30.0) # sleeps between the 5 attempts (spec)
|
||||
MAX_ATTEMPTS = 5
|
||||
_UNSAFE = re.compile(r"[^A-Za-z0-9._=,-]+")
|
||||
|
||||
|
||||
class BGGQueueTimeout(Exception):
|
||||
"""BGG kept answering 202 (or throttling) past the retry budget."""
|
||||
|
||||
|
||||
def cache_key(endpoint: str, params: dict[str, str]) -> str:
|
||||
query = urlencode(sorted(params.items()))
|
||||
digest = hashlib.md5(f"{endpoint}?{query}".encode()).hexdigest()[:10]
|
||||
slug = _UNSAFE.sub("-", query)[:80].strip("-")
|
||||
return f"{endpoint}_{slug}_{digest}.xml"
|
||||
|
||||
|
||||
class BGGClient:
|
||||
def __init__(
|
||||
self,
|
||||
cache_dir: Path,
|
||||
rate_limit_seconds: float = 2.0,
|
||||
transport: httpx.BaseTransport | None = None,
|
||||
sleep: Callable[[float], None] = time.sleep,
|
||||
monotonic: Callable[[], float] = time.monotonic,
|
||||
rng: random.Random | None = None,
|
||||
) -> None:
|
||||
self.cache_dir = cache_dir
|
||||
self._rate = rate_limit_seconds
|
||||
self._sleep = sleep
|
||||
self._monotonic = monotonic
|
||||
self._rng = rng or random.Random()
|
||||
self._last_request: float | None = None
|
||||
self._http = httpx.Client(
|
||||
base_url=BASE_URL,
|
||||
timeout=30.0,
|
||||
headers={"User-Agent": "bggpipe/0.1 (shelf-collection pipeline)"},
|
||||
transport=transport,
|
||||
)
|
||||
|
||||
def _throttle(self) -> None:
|
||||
if self._last_request is not None:
|
||||
wait = self._rate - (self._monotonic() - self._last_request)
|
||||
if wait > 0:
|
||||
self._sleep(wait)
|
||||
|
||||
def get_xml(self, endpoint: str, params: dict[str, str]) -> str:
|
||||
"""Fetch one endpoint, serving from and filling the disk cache."""
|
||||
cache_path = self.cache_dir / cache_key(endpoint, params)
|
||||
if cache_path.exists():
|
||||
return cache_path.read_text()
|
||||
|
||||
for attempt in range(MAX_ATTEMPTS):
|
||||
self._throttle()
|
||||
response = self._http.get(f"/{endpoint}", params=params)
|
||||
self._last_request = self._monotonic()
|
||||
|
||||
if response.status_code == 202:
|
||||
if attempt < MAX_ATTEMPTS - 1:
|
||||
self._sleep(QUEUE_BACKOFF[min(attempt, len(QUEUE_BACKOFF) - 1)])
|
||||
continue
|
||||
if response.status_code in (429, 503):
|
||||
if attempt < MAX_ATTEMPTS - 1:
|
||||
backoff = 2.0 * (2**attempt) * (1 + self._rng.uniform(0, 0.5))
|
||||
self._sleep(backoff)
|
||||
continue
|
||||
|
||||
response.raise_for_status()
|
||||
self.cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
cache_path.write_text(response.text)
|
||||
return response.text
|
||||
|
||||
raise BGGQueueTimeout(
|
||||
f"BGG did not serve /{endpoint} after {MAX_ATTEMPTS} attempts "
|
||||
"(still queued or throttled) — wait a minute and re-run; "
|
||||
"completed work is cached."
|
||||
)
|
||||
|
||||
# -- typed endpoint wrappers ------------------------------------------
|
||||
|
||||
def search(
|
||||
self, query: str, types: str = "boardgame,boardgameexpansion"
|
||||
) -> list[SearchResult]:
|
||||
return parse_search(self.get_xml("search", {"query": query, "type": types}))
|
||||
|
||||
def things(
|
||||
self,
|
||||
ids: Iterable[int],
|
||||
stats: bool = False,
|
||||
versions: bool = False,
|
||||
) -> list[ThingDetails]:
|
||||
params = {"id": ",".join(str(i) for i in ids)}
|
||||
if stats:
|
||||
params["stats"] = "1"
|
||||
if versions:
|
||||
params["versions"] = "1"
|
||||
return parse_things(self.get_xml("thing", params))
|
||||
|
||||
def collection(
|
||||
self,
|
||||
username: str,
|
||||
subtype: str | None = None,
|
||||
version: bool = True,
|
||||
) -> list[CollectionItem]:
|
||||
params = {"username": username, "own": "1"}
|
||||
if subtype:
|
||||
params["subtype"] = subtype
|
||||
if version:
|
||||
params["version"] = "1"
|
||||
return parse_collection(self.get_xml("collection", params))
|
||||
|
||||
def collection_full(self, username: str) -> list[CollectionItem]:
|
||||
"""Owned items incl. expansions (excluded from the default subtype)."""
|
||||
base = self.collection(username)
|
||||
expansions = self.collection(username, subtype="boardgameexpansion")
|
||||
seen = {item.coll_id for item in base}
|
||||
return base + [e for e in expansions if e.coll_id not in seen]
|
||||
@@ -0,0 +1,154 @@
|
||||
"""Dataclasses + XML parsers for the BGG XML API2 responses we consume."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import xml.etree.ElementTree as ET # element types only; parsing goes via defusedxml
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from defusedxml.ElementTree import fromstring as _safe_fromstring
|
||||
|
||||
|
||||
class BGGResponseError(Exception):
|
||||
"""The API returned a well-formed error document (e.g. bad username)."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SearchResult:
|
||||
bgg_id: int
|
||||
name: str
|
||||
name_type: str # "primary" | "alternate"
|
||||
year: int | None
|
||||
type: str # "boardgame" | "boardgameexpansion"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class GameVersion:
|
||||
version_id: int
|
||||
name: str
|
||||
year: int | None
|
||||
publishers: tuple[str, ...]
|
||||
languages: tuple[str, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ThingDetails:
|
||||
bgg_id: int
|
||||
name: str
|
||||
year: int | None
|
||||
type: str
|
||||
owned: int | None = None
|
||||
rank: int | None = None
|
||||
versions: tuple[GameVersion, ...] = field(default=())
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CollectionItem:
|
||||
object_id: int
|
||||
coll_id: int
|
||||
name: str
|
||||
subtype: str
|
||||
own: bool
|
||||
year: int | None
|
||||
version_id: int | None
|
||||
|
||||
|
||||
def _root(xml_text: str) -> ET.Element:
|
||||
root = _safe_fromstring(xml_text)
|
||||
if root.tag == "errors":
|
||||
message = root.findtext("./error/message") or "unknown BGG error"
|
||||
raise BGGResponseError(message.strip())
|
||||
return root
|
||||
|
||||
|
||||
def _attr_int(elem: ET.Element | None, attr: str = "value") -> int | None:
|
||||
if elem is None:
|
||||
return None
|
||||
raw = elem.get(attr)
|
||||
if raw is None or not raw.lstrip("-").isdigit():
|
||||
return None
|
||||
return int(raw)
|
||||
|
||||
|
||||
def parse_search(xml_text: str) -> list[SearchResult]:
|
||||
results = []
|
||||
for item in _root(xml_text).findall("item"):
|
||||
name_elem = item.find("name")
|
||||
if name_elem is None or item.get("id") is None:
|
||||
continue
|
||||
results.append(
|
||||
SearchResult(
|
||||
bgg_id=int(item.get("id")),
|
||||
name=name_elem.get("value", ""),
|
||||
name_type=name_elem.get("type", "primary"),
|
||||
year=_attr_int(item.find("yearpublished")),
|
||||
type=item.get("type", "boardgame"),
|
||||
)
|
||||
)
|
||||
return results
|
||||
|
||||
|
||||
def _parse_version(item: ET.Element) -> GameVersion | None:
|
||||
version_id = item.get("id")
|
||||
name = item.find("name[@type='primary']")
|
||||
if version_id is None:
|
||||
return None
|
||||
return GameVersion(
|
||||
version_id=int(version_id),
|
||||
name=name.get("value", "") if name is not None else "",
|
||||
year=_attr_int(item.find("yearpublished")),
|
||||
publishers=tuple(
|
||||
link.get("value", "")
|
||||
for link in item.findall("link[@type='boardgamepublisher']")
|
||||
),
|
||||
languages=tuple(
|
||||
link.get("value", "") for link in item.findall("link[@type='language']")
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def parse_things(xml_text: str) -> list[ThingDetails]:
|
||||
things = []
|
||||
for item in _root(xml_text).findall("item"):
|
||||
name = item.find("name[@type='primary']")
|
||||
rank_elem = item.find(".//ranks/rank[@name='boardgame']")
|
||||
versions = [
|
||||
v
|
||||
for v_item in item.findall("versions/item")
|
||||
if (v := _parse_version(v_item)) is not None
|
||||
]
|
||||
things.append(
|
||||
ThingDetails(
|
||||
bgg_id=int(item.get("id", 0)),
|
||||
name=name.get("value", "") if name is not None else "",
|
||||
year=_attr_int(item.find("yearpublished")),
|
||||
type=item.get("type", "boardgame"),
|
||||
owned=_attr_int(item.find(".//ratings/owned")),
|
||||
rank=_attr_int(rank_elem),
|
||||
versions=tuple(versions),
|
||||
)
|
||||
)
|
||||
return things
|
||||
|
||||
|
||||
def parse_collection(xml_text: str) -> list[CollectionItem]:
|
||||
items = []
|
||||
for item in _root(xml_text).findall("item"):
|
||||
status = item.find("status")
|
||||
year_text = item.findtext("yearpublished")
|
||||
version_item = item.find("version/item")
|
||||
items.append(
|
||||
CollectionItem(
|
||||
object_id=int(item.get("objectid", 0)),
|
||||
coll_id=int(item.get("collid", 0)),
|
||||
name=item.findtext("name", default=""),
|
||||
subtype=item.get("subtype", "boardgame"),
|
||||
own=status is not None and status.get("own") == "1",
|
||||
year=int(year_text) if year_text and year_text.isdigit() else None,
|
||||
version_id=(
|
||||
int(version_item.get("id"))
|
||||
if version_item is not None and version_item.get("id")
|
||||
else None
|
||||
),
|
||||
)
|
||||
)
|
||||
return items
|
||||
Reference in New Issue
Block a user