How to Automate Shopify Inventory Updates using the API

1import os
2import csv
3import time
4import logging
5import requests
6
7logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
8
9SHOP_DOMAIN = os.environ['SHOPIFY_SHOP_DOMAIN']
10ACCESS_TOKEN = os.environ['SHOPIFY_ACCESS_TOKEN']
11LOCATION_ID = os.environ['SHOPIFY_LOCATION_ID']
12SUPPLIER_FEED_PATH = os.environ.get('SUPPLIER_FEED_PATH', 'supplier_feed.csv')
13BATCH_SIZE = 50
14
15URL = f'https://{SHOP_DOMAIN}/admin/api/2026-04/graphql.json'
16HEADERS = {'X-Shopify-Access-Token': ACCESS_TOKEN, 'Content-Type': 'application/json'}
17
18def graphql(query, variables=None):
19    resp = requests.post(URL, json={'query': query, 'variables': variables or {}}, headers=HEADERS)
20    resp.raise_for_status()
21    return resp.json()
22
23def fetch_inventory_map():
24    """Returns {sku: {item_id, available}}"""
25    inventory = {}
26    cursor = None
27    while True:
28        after = f', after: "{cursor}"' if cursor else ''
29        q = f'{{ location(id: "{LOCATION_ID}") {{ inventoryLevels(first: 250{after}) {{ edges {{ node {{ available item {{ id sku }} }} }} pageInfo {{ hasNextPage endCursor }} }} }} }}'
30        data = graphql(q)['data']['location']['inventoryLevels']
31        for edge in data['edges']:
32            n = edge['node']
33            inventory[n['item']['sku']] = {'item_id': n['item']['id'], 'available': n['available']}
34        if not data['pageInfo']['hasNextPage']:
35            break
36        cursor = data['pageInfo']['endCursor']
37    return inventory
38
39def parse_feed(path):
40    feed = {}
41    with open(path) as f:
42        for row in csv.DictReader(f):
43            feed[row['SKU']] = int(row['quantity'])
44    return feed
45
46def adjust_batch(changes):
47    mutation = 'mutation inventoryAdjustQuantities($input: InventoryAdjustQuantitiesInput!) { inventoryAdjustQuantities(input: $input) { userErrors { field message } inventoryAdjustmentGroup { changes { name delta quantityAfterChange } } } }'
48    variables = {'input': {'reason': 'correction', 'name': 'available', 'changes': changes}}
49    result = graphql(mutation, variables)
50    cost = result.get('extensions', {}).get('cost', {}).get('throttleStatus', {})
51    if cost.get('currentlyAvailable', 1000) < 150:
52        logging.info('Throttle low, sleeping 3s')
53        time.sleep(3)
54    errors = result['data']['inventoryAdjustQuantities']['userErrors']
55    if errors:
56        logging.error(f'User errors: {errors}')
57    return result
58
59def main():
60    logging.info('Fetching Shopify inventory map...')
61    inventory_map = fetch_inventory_map()
62    logging.info(f'Found {len(inventory_map)} SKUs in Shopify')
63
64    logging.info(f'Parsing supplier feed from {SUPPLIER_FEED_PATH}...')
65    feed = parse_feed(SUPPLIER_FEED_PATH)
66
67    changes = []
68    for sku, supplier_qty in feed.items():
69        if sku not in inventory_map:
70            logging.warning(f'SKU {sku} not found in Shopify, skipping')
71            continue
72        delta = supplier_qty - inventory_map[sku]['available']
73        if delta != 0:
74            changes.append({'inventoryItemId': inventory_map[sku]['item_id'], 'locationId': LOCATION_ID, 'delta': delta})
75
76    logging.info(f'{len(changes)} SKUs require adjustment')
77    for i in range(0, len(changes), BATCH_SIZE):
78        batch = changes[i:i + BATCH_SIZE]
79        adjust_batch(batch)
80        logging.info(f'Processed batch {i // BATCH_SIZE + 1}')
81
82    logging.info('Inventory sync complete')
83
84if __name__ == '__main__':
85    main()