set_deny_on_zero_stock.py

set_deny_on_zero_stock.py

import requests
import time

SHOP_NAME = "smepay-dev"
ACCESS_TOKEN = "shpat_****143aa466e90046d79dcbe3d6****"
API_VERSION = "2023-07"

headers = {
    "X-Shopify-Access-Token": ACCESS_TOKEN,
    "Content-Type": "application/json"
}

def get_all_products():
    products = []
    url = f"https://{SHOP_NAME}.myshopify.com/admin/api/{API_VERSION}/products.json?limit=250"
    page = 1

    while url:
        print(f"📦 Fetching product page {page}...")
        resp = requests.get(url, headers=headers)

        if resp.status_code == 429:
            print("⚠️ Rate limit hit while fetching products. Sleeping for 5 seconds...")
            time.sleep(5)
            continue

        resp.raise_for_status()
        data = resp.json()
        products_batch = data.get("products", [])
        print(f"✅ Fetched {len(products_batch)} products from page {page}")
        products.extend(products_batch)

        # Delay to respect rate limit
        time.sleep(0.5)

        # Pagination
        link = resp.headers.get("Link")
        if link and 'rel="next"' in link:
            next_url = [l.split(";")[0].strip("<> ") for l in link.split(",") if 'rel="next"' in l][0]
            url = next_url
            page += 1
        else:
            url = None
    return products

def update_variant_inventory_policy(variant_id):
    url = f"https://{SHOP_NAME}.myshopify.com/admin/api/{API_VERSION}/variants/{variant_id}.json"
    data = {
        "variant": {
            "inventory_policy": "deny"
        }
    }

    resp = requests.put(url, headers=headers, json=data)

    if resp.status_code == 429:
        print(f"⚠️ Rate limit hit while updating variant {variant_id}. Sleeping for 5 seconds...")
        time.sleep(5)
        return update_variant_inventory_policy(variant_id)  # Retry once

    if resp.status_code == 200:
        print(f"✅ Updated variant {variant_id} to inventory_policy = deny")
    else:
        print(f"❌ Failed to update variant {variant_id}: {resp.status_code} — {resp.text}")

    # Delay to respect rate limit
    time.sleep(0.5)

def main():
    products = get_all_products()
    print(f"\n🎯 Total products fetched: {len(products)}\n")

    for product in products:
        product_id = product["id"]
        product_title = product.get("title", "Untitled")

        for variant in product["variants"]:
            variant_id = variant["id"]
            inventory_qty = variant.get("inventory_quantity", 0)
            current_policy = variant.get("inventory_policy", "unknown")

            print(f"🔍 Product: '{product_title}' | Variant ID: {variant_id} | Inventory: {inventory_qty} | Policy: {current_policy}")

            if inventory_qty <= 0:
                if current_policy != "deny":
                    print(f"➡️  Inventory is {inventory_qty} and policy is '{current_policy}' — updating to 'deny'")
                    update_variant_inventory_policy(variant_id)
                else:
                    print(f"✅ Already set to 'deny', skipping")
            else:
                print(f"🚫 Inventory is {inventory_qty} — skipping (stock available)\n")

if __name__ == "__main__":
    main()