Get low-stock warnings from Shopify in Slack

from datetime import date, timedelta
from collections import Counter
from clickruntime import on_trigger, credentials, http, notify, state

STORE = "your-store.myshopify.com"

def gql(query, variables=None):
    resp = http.post(f"https://{STORE}/admin/api/2025-07/graphql.json", json={"query": query, "variables": variables or {}},
                     headers={"X-Shopify-Access-Token": credentials.connection("shopify", "store")})
    return resp.raise_for_status() or resp.json()["data"]

DAYS = 14

@on_trigger
def run(event):
    sold, after = Counter(), None
    while True:
        page = gql("""query($q: String!, $a: String) { orders(first: 100, after: $a, query: $q) { pageInfo { hasNextPage endCursor }
            nodes { lineItems(first: 50) { nodes { quantity variant { id } } } } } }""", {"q": f"created_at:>={date.today() - timedelta(days=30)}", "a": after})["orders"]
        sold.update({li["variant"]["id"]: li["quantity"] for o in page["nodes"] for li in o["lineItems"]["nodes"] if li["variant"]})
        if not page["pageInfo"]["hasNextPage"]:
            break
        after = page["pageInfo"]["endCursor"]
    variants = gql("{ productVariants(first: 250) { nodes { id displayName inventoryQuantity } } }")["productVariants"]["nodes"]
    flagged, low = set(state.get("flagged") or []), []
    for v in variants:
        days = v["inventoryQuantity"] / (sold[v["id"]] / 30) if sold[v["id"]] else None
        if days is not None and days < DAYS:
            if v["id"] not in flagged:
                low.append((days, f"{v['displayName']}: {v['inventoryQuantity']} left, about {days:.0f} days"))
            flagged.add(v["id"])
        else:
            flagged.discard(v["id"])
    if low:
        notify.slack("Running low in Shopify\n" + "\n".join(t for _, t in sorted(low)), channel="#ops")
    state.set("flagged", sorted(flagged))
    return {"new_low": len(low)}