fetch.py

1import json2import os3import time4from catalog.config import DELAY, PRINTFUL_API_KEY, PRINTFUL_URL5from catalog.helpers import ROOT6from env import load_json, save_json7from urllib.request import Request, urlopen89HEADERS = {"Authorization": f"Bearer {PRINTFUL_API_KEY}"}1011RAW = os.path.join(ROOT, "data/raw")1213def raw_path(kind: str, id: int = None) -> str:14    if id is None:15        return os.path.join(RAW, f"{kind}.json")16    return os.path.join(RAW, kind, f"{id}.json")1718def get(url: str) -> dict:19    with urlopen(Request(url, headers=HEADERS), timeout=30) as response:20        return json.loads(response.read())2122def get_paged(url: str, key: str = None) -> dict:23    data = get(url)24    rows = data["data"] if key is None else data["data"][key]25    while "next" in data.get("_links", {}):26        time.sleep(DELAY)27        data = get(data["_links"]["next"]["href"])28        rows.extend(data["data"] if key is None else data["data"][key])29    if key is None:30        data["data"] = rows31    else:32        data["data"][key] = rows33    return data3435# CATALOG3637def fetch_catalog() -> list[dict]:38    print("Fetching the Printful catalog")39    data = get_paged(f"{PRINTFUL_URL}/v2/catalog-products?limit=100")40    rows = data["data"]41    save_json(raw_path("catalog"), rows)42    print(f"Fetched {len(rows)} catalog rows")43    return rows4445def load_raw_catalog(refetch: bool = False) -> list[dict]:46    path = raw_path("catalog")47    if refetch or not os.path.exists(path):48        return fetch_catalog()49    rows = load_json(path)50    print(f"Reusing {len(rows)} catalog rows from data/raw/catalog.json")51    return rows5253# PRODUCTS5455def fetch_one(kind: str, url: str, id: int, paged: bool, key: str = None) -> None:56    data = get_paged(url, key) if paged else get(url)57    save_json(raw_path(kind, id), data)58    print(f"Fetched {kind} for product {id}")59    time.sleep(DELAY)6061def fetch_product(id: int) -> None:62    base = f"{PRINTFUL_URL}/v2/catalog-products/{id}"63    fetch_one("products", base, id, False)64    fetch_one("variants", f"{base}/catalog-variants?limit=100", id, True)65    fetch_one("prices", f"{base}/prices?limit=100", id, True, "variants")66    fetch_one("mockups", f"{base}/mockup-styles?limit=100", id, False)6768def has_raw(id: int) -> bool:69    kinds = ["products", "variants", "prices", "mockups"]70    return all(os.path.exists(raw_path(kind, id)) for kind in kinds)7172def fetch_products(ids: list[int], refetch: bool = False) -> None:73    print(f"Fetching product data for {len(ids)} products")74    for id in ids:75        if not refetch and has_raw(id):76            print(f"Reusing raw data for product {id}")77            continue78        fetch_product(id)79    print(f"Fetched product data for {len(ids)} products")