shopify.py

1import json2import urllib.error3import urllib.parse4import urllib.request56from env import gate, load_env, need, save_env, say, verb78load_env()910# API1112API_VERSION = "2026-07"13SHOP = need("SHOPIFY_SHOP_URL")14ADMIN_URL = f"https://{SHOP}/admin/api/{API_VERSION}/graphql.json"15AUTH_URL = f"https://{SHOP}/admin/oauth/access_token"16ONLINE_STORE = "Online Store"17HEADLESS = "Headless"1819def post(url, body, headers):20    request = urllib.request.Request(url, data=body, headers=headers, method="POST")21    try:22        with urllib.request.urlopen(request, timeout=30) as response:23            return json.loads(response.read())24    except urllib.error.HTTPError as error:25        raise SystemExit(f"admin api {error.code}")2627def token():28    body = urllib.parse.urlencode({29        "client_id": need("SHOPIFY_CLIENT_ID"),30        "client_secret": need("SHOPIFY_SECRET"),31        "grant_type": "client_credentials",32    }).encode()33    headers = {"Content-Type": "application/x-www-form-urlencoded"}34    data = post(AUTH_URL, body, headers)35    if "access_token" not in data:36        raise SystemExit("admin token refused")37    return data["access_token"]3839def graphql(access, query, variables=None):40    body = json.dumps({"query": query, "variables": variables or {}}).encode()41    headers = {"Content-Type": "application/json", "X-Shopify-Access-Token": access}42    data = post(ADMIN_URL, body, headers)43    if "errors" in data:44        raise SystemExit(f"graphql: {json.dumps(data['errors'])[:300]}")45    return data["data"]4647# QUERIES4849PUBLICATIONS = """50query {51    publications(first: 50) {52        nodes { id name }53    }54}55"""5657PRODUCTS = """58query($cursor: String) {59    products(first: 250, after: $cursor) {60        pageInfo { hasNextPage endCursor }61        nodes { id title }62    }63}64"""6566DELETE = """67mutation($input: ProductDeleteInput!) {68    productDelete(input: $input) {69        deletedProductId70        userErrors { field message }71    }72}73"""7475# PUBLICATIONS7677def publications():78    if not gate("publications", [79        f"admin graphql {API_VERSION} on the shop behind SHOPIFY_SHOP_URL",80        f"list every publication, find {ONLINE_STORE} and {HEADLESS}",81        "save SHOPIFY_ONLINE_STORE_ID and SHOPIFY_HEADLESS_ID",82    ]): return83    nodes = graphql(token(), PUBLICATIONS)["publications"]["nodes"]84    store = ""85    headless = ""86    for node in nodes:87        say(f"  {node['name']}")88        if node["name"] == ONLINE_STORE: store = node["id"]89        if HEADLESS in node["name"]: headless = node["id"]90    if not store:91        raise SystemExit(f"refuse: no {ONLINE_STORE} publication on this shop")92    if not headless:93        raise SystemExit(f"refuse: no {HEADLESS} publication on this shop")94    save_env("SHOPIFY_ONLINE_STORE_ID", store)95    save_env("SHOPIFY_HEADLESS_ID", headless)9697# PURGE9899def purge():100    access = token()101    ids = []102    cursor = None103    while True:104        page = graphql(access, PRODUCTS, {"cursor": cursor})["products"]105        ids += [node["id"] for node in page["nodes"]]106        if not page["pageInfo"]["hasNextPage"]: break107        cursor = page["pageInfo"]["endCursor"]108    if not gate("purge", [109        f"delete {len(ids)} products from the shop behind SHOPIFY_SHOP_URL",110        "productDelete, one call each, no undo",111    ]): return112    for count, one in enumerate(ids, 1):113        result = graphql(access, DELETE, {"input": {"id": one}})["productDelete"]114        if result["userErrors"]:115            say(f"  {one} {result['userErrors']}")116            continue117        say(f"  deleted {count}/{len(ids)}")118    say(f"{len(ids)} products deleted")119120# MAIN121122VERBS = {123    "publications": publications,124    "purge": purge,125}126127if __name__ == "__main__":128    VERBS[verb(list(VERBS))]()