Reconciler Inventory
Bulk stock true-up after a bad import
Someone uploaded a stock file, and now half the catalog shows the wrong number on hand. Maybe the wrong column got mapped to quantity, maybe an old export got uploaded by mistake, maybe a sync app ran twice. Whatever happened, Shopify does not know the file was bad, it just wrote down whatever the file said. Here is why a bad import spreads wrong counts across hundreds of items at once, and a small script that compares the live store against a trusted snapshot and re-sets only what is actually wrong, in safe batches.
A bulk import writes an absolute quantity for every row in the file, so a bad file leaves a bad number on every item it touched, and there is no undo button for that. Run a small Python or Node.js script that loads a trusted snapshot of what the counts should be (the export you took right before the import, or your warehouse system's numbers), reads the live available quantity for each item from Shopify, and only corrects the items where the drift is real and inside a sane size limit. It writes with inventorySetQuantities using compareQuantity, in small batches, so nothing gets overwritten out from under a sale that happens mid-run. Full code, tests, and a dry run guard are below.
The problem in plain words
A bulk stock update, whether it comes from a CSV upload, a spreadsheet tool, or a third party sync app, almost always writes an absolute quantity. It does not ask "is this different from before," it just says "set this item's available count to this number." That is exactly what you want when the file is right.
The trouble starts when the file is wrong. A column got shifted one to the left, so prices landed in the quantity field. An old export from last month got re-uploaded by accident. A sync app double ran and pushed a warehouse count that was never adjusted for local sales. In every case, Shopify does precisely what it was told, and the wrong number becomes the new truth for every item the file mentioned. There is no built-in way to tell Shopify "that whole file was garbage, please undo it," because Shopify never saw a file, it only saw a series of writes.
Why it happens
Bulk tools are built to overwrite, not to merge, because that is what makes them fast for a legitimate stock take. A few common ways this turns into a mess:
- A spreadsheet export mapped the wrong column to quantity, so prices, weights, or SKU numbers land in the available field.
- An old export gets re-uploaded by mistake, quietly rolling every item back to a count from weeks ago.
- A third party inventory sync app runs twice, or runs against a warehouse feed that was never adjusted for sales that happened since the feed was built.
- A migration or bulk edit tool applies to more locations or more products than intended, because a filter was missed.
This is a common source of panic. The moment someone notices, the instinct is to fix it fast, but fixing it fast without a plan usually means guessing at numbers or re-running another bulk write, which risks making things worse. The safest path is to compare the store against something you trust, and only touch what is actually wrong. See the citations at the end for the exact docs.
You cannot undo a bad import, but you can true it up, if you have a trusted snapshot of what the counts should be, taken before the bad file landed. The safe pattern is not "reset everything to some default." It is "compare live counts to the snapshot, correct only real drift, and skip anything the drift looks too large to trust." A swing far bigger than normal usually means the snapshot itself is stale for that item, not that the item truly lost that much stock, so it should go to a human instead of being auto corrected.
The fix, as a flow
We do not touch products or prices, only the available quantity at one location. The job reads a small CSV snapshot you trust, walks the live inventory levels at that location, and for every SKU it can compare, works out whether the drift is real and inside a sane size limit. Anything outside the snapshot, already correct, or suspiciously large is left alone. Everything else gets corrected with a compare-and-set write, in small batches so a single bad batch cannot take down the whole run.
Build it step by step
Get an Admin API access token and a trusted snapshot
Create a custom app in your Shopify admin under Settings, Apps and sales channels, Develop apps. Give it the read_inventory and write_inventory scopes and install it to get an Admin API access token that starts with shpat_. Then find the trusted snapshot, the export you took right before the bad import, or a warehouse system's own counts, and save it as a small CSV with a sku and an available column. Keep the token and the shop domain in environment variables, never in the file.
pip install requests
export SHOPIFY_SHOP="yourstore.myshopify.com"
export SHOPIFY_ACCESS_TOKEN="shpat_..."
export SHOPIFY_LOCATION_ID="gid://shopify/Location/123456789"
export SNAPSHOT_PATH="snapshot.csv" # columns: sku,available
export MAX_ADJUST="500"
export BATCH_SIZE="25"
export DRY_RUN="true" # start safe, change to false to write// Node 18+ has fetch built in, no dependencies needed
export SHOPIFY_SHOP="yourstore.myshopify.com"
export SHOPIFY_ACCESS_TOKEN="shpat_..."
export SHOPIFY_LOCATION_ID="gid://shopify/Location/123456789"
export SNAPSHOT_PATH="snapshot.csv" // columns: sku,available
export MAX_ADJUST="500"
export BATCH_SIZE="25"
export DRY_RUN="true" // start safe, change to false to writeTalk to the Admin GraphQL API
Every call goes to one GraphQL endpoint with your token in the X-Shopify-Access-Token header. A small helper sends a query and returns the data, and raises if Shopify reports an error. We use this same helper to read live inventory levels and to run the correction mutation.
import os, requests
SHOP = os.environ["SHOPIFY_SHOP"]
TOKEN = os.environ["SHOPIFY_ACCESS_TOKEN"]
API_VERSION = os.environ.get("SHOPIFY_API_VERSION", "2025-01")
ENDPOINT = f"https://{SHOP}/admin/api/{API_VERSION}/graphql.json"
def gql(query, variables=None):
r = requests.post(
ENDPOINT,
json={"query": query, "variables": variables or {}},
headers={"X-Shopify-Access-Token": TOKEN, "Content-Type": "application/json"},
timeout=30,
)
r.raise_for_status()
body = r.json()
if body.get("errors"):
raise RuntimeError(body["errors"])
return body["data"]const SHOP = process.env.SHOPIFY_SHOP;
const TOKEN = process.env.SHOPIFY_ACCESS_TOKEN;
const API_VERSION = process.env.SHOPIFY_API_VERSION || "2025-01";
const ENDPOINT = `https://${SHOP}/admin/api/${API_VERSION}/graphql.json`;
async function gql(query, variables = {}) {
const res = await fetch(ENDPOINT, {
method: "POST",
headers: { "X-Shopify-Access-Token": TOKEN, "Content-Type": "application/json" },
body: JSON.stringify({ query, variables }),
});
if (!res.ok) throw new Error(`Shopify ${res.status}`);
const body = await res.json();
if (body.errors) throw new Error(JSON.stringify(body.errors));
return body.data;
}Read the live counts for the location
Ask for every inventory level at the affected location, with the available quantity and the item's SKU. We page through with a cursor so the job handles a large catalog without missing anything.
LOCATION_LEVELS_QUERY = """
query($id: ID!, $cursor: String) {
location(id: $id) {
inventoryLevels(first: 100, after: $cursor) {
pageInfo { hasNextPage endCursor }
nodes {
item { id sku }
quantities(names: ["available"]) { name quantity }
}
}
}
}"""
def live_levels(location_id):
cursor = None
while True:
data = gql(LOCATION_LEVELS_QUERY, {"id": location_id, "cursor": cursor})["location"]
levels = data["inventoryLevels"]
for node in levels["nodes"]:
available = next(q["quantity"] for q in node["quantities"] if q["name"] == "available")
yield node["item"]["id"], node["item"]["sku"], available
if not levels["pageInfo"]["hasNextPage"]:
return
cursor = levels["pageInfo"]["endCursor"]const LOCATION_LEVELS_QUERY = `
query($id: ID!, $cursor: String) {
location(id: $id) {
inventoryLevels(first: 100, after: $cursor) {
pageInfo { hasNextPage endCursor }
nodes {
item { id sku }
quantities(names: ["available"]) { name quantity }
}
}
}
}`;
async function* liveLevels(locationId) {
let cursor = null;
while (true) {
const data = (await gql(LOCATION_LEVELS_QUERY, { id: locationId, cursor })).location;
const levels = data.inventoryLevels;
for (const node of levels.nodes) {
const available = node.quantities.find((q) => q.name === "available").quantity;
yield { itemId: node.item.id, sku: node.item.sku, available };
}
if (!levels.pageInfo.hasNextPage) return;
cursor = levels.pageInfo.endCursor;
}
}Decide, with one pure function
Keep the decision in its own function that takes a SKU, the live quantity, the snapshot, and a size guard, and returns either a correction or nothing. A pure function like this is easy to read and easy to test, which we do later. The rule is strict on purpose. A SKU with no entry in the snapshot is left alone, since there is nothing trusted to restore it to. A SKU that already matches is left alone. And a drift larger than the guard is left alone too, because a swing that big usually means the snapshot is stale for that item, not that the item truly lost that much stock, so it should go to a human instead of being applied automatically.
def plan_correction(sku, live_quantity, snapshot, max_adjust):
if sku not in snapshot:
return None
target = snapshot[sku]
delta = target - live_quantity
if delta == 0:
return None
if abs(delta) > max_adjust:
return None
return {"sku": sku, "from": live_quantity, "to": target, "delta": delta}export function planCorrection(sku, liveQuantity, snapshot, maxAdjust) {
if (!(sku in snapshot)) return null;
const target = snapshot[sku];
const delta = target - liveQuantity;
if (delta === 0) return null;
if (Math.abs(delta) > maxAdjust) return null;
return { sku, from: liveQuantity, to: target, delta };
}Write corrections in small compare-and-set batches
When a SKU is eligible, queue it into a batch and call inventorySetQuantities with the target quantity as the new value and the live quantity you just read as compareQuantity. That makes the write a compare-and-set. Shopify only applies it if the persisted quantity still matches what you read a moment ago, so a sale that lands between your read and your write is rejected, not silently overwritten. Batches keep any single mutation call small, so one bad batch never risks the whole run, and a short pause between batches keeps the API happy.
SET_QUANTITIES_MUTATION = """
mutation($input: InventorySetQuantitiesInput!) {
inventorySetQuantities(input: $input) {
inventoryAdjustmentGroup { reason changes { name delta quantityAfterChange } }
userErrors { field message }
}
}"""
def apply_batch(location_id, corrections_by_item):
quantities = [
{
"inventoryItemId": item_id,
"locationId": location_id,
"quantity": target,
"compareQuantity": live_quantity,
}
for item_id, live_quantity, target in corrections_by_item
]
result = gql(
SET_QUANTITIES_MUTATION,
{"input": {"name": "available", "reason": "correction",
"ignoreCompareQuantity": False, "quantities": quantities}},
)["inventorySetQuantities"]
if result["userErrors"]:
raise RuntimeError(result["userErrors"])
return result["inventoryAdjustmentGroup"]const SET_QUANTITIES_MUTATION = `
mutation($input: InventorySetQuantitiesInput!) {
inventorySetQuantities(input: $input) {
inventoryAdjustmentGroup { reason changes { name delta quantityAfterChange } }
userErrors { field message }
}
}`;
async function applyBatch(locationId, corrections) {
const quantities = corrections.map(({ itemId, liveQuantity, target }) => ({
inventoryItemId: itemId,
locationId,
quantity: target,
compareQuantity: liveQuantity,
}));
const result = (await gql(SET_QUANTITIES_MUTATION, {
input: { name: "available", reason: "correction", ignoreCompareQuantity: false, quantities },
})).inventorySetQuantities;
if (result.userErrors.length) throw new Error(JSON.stringify(result.userErrors));
return result.inventoryAdjustmentGroup;
}Wire it together with a dry run guard
The loop ties every piece together. Notice the dry run guard. On the first run, leave DRY_RUN on so the script only lists which SKUs it would correct and by how much. Read the output, agree with it, then switch it off to let it write. It runs once, on demand, right after you confirm the snapshot is trustworthy, not on a schedule.
Always start with DRY_RUN=true, and only trust a snapshot you know predates the bad import. A snapshot that is itself stale will just replace one wrong number with another, so double check where it came from before the first real run.
The full code
Here is the complete script in one file for each language. It reads settings from the environment, loads the snapshot, logs what it does, respects the dry run flag, and is safe to run again and again because it only touches SKUs the snapshot covers, skips anything already correct, skips drift outside the size guard, and writes with a compare-and-set check.
View this code on GitHub Full runnable folder with tests in the shopify-fixes repo.
"""True up Shopify available inventory after a bad bulk import, safely.
A bad import (a bad CSV, a stuck sync app) can stamp the wrong "available"
quantity onto many items at once. This script reads a trusted snapshot (the
counts you captured before the bad import, keyed by SKU and location), reads
each item's live quantity from Shopify, and only corrects items where the
drift is real and inside a sane guard. It writes with inventorySetQuantities
using compareQuantity, so a sale that lands between the read and the write
is never silently overwritten. Batched, paged, and safe to run again and
again.
"""
import csv
import logging
import os
import time
import requests
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("true_up_inventory")
SHOP = os.environ["SHOPIFY_SHOP"]
TOKEN = os.environ["SHOPIFY_ACCESS_TOKEN"]
API_VERSION = os.environ.get("SHOPIFY_API_VERSION", "2025-01")
ENDPOINT = f"https://{SHOP}/admin/api/{API_VERSION}/graphql.json"
LOCATION_ID = os.environ.get("SHOPIFY_LOCATION_ID", "gid://shopify/Location/0")
SNAPSHOT_PATH = os.environ.get("SNAPSHOT_PATH", "snapshot.csv")
MAX_ADJUST = int(os.environ.get("MAX_ADJUST", "500"))
BATCH_SIZE = int(os.environ.get("BATCH_SIZE", "25"))
DRY_RUN = os.environ.get("DRY_RUN", "true").lower() == "true"
LOCATION_LEVELS_QUERY = """
query($id: ID!, $cursor: String) {
location(id: $id) {
inventoryLevels(first: 100, after: $cursor) {
pageInfo { hasNextPage endCursor }
nodes {
item { id sku }
quantities(names: ["available"]) { name quantity }
}
}
}
}"""
SET_QUANTITIES_MUTATION = """
mutation($input: InventorySetQuantitiesInput!) {
inventorySetQuantities(input: $input) {
inventoryAdjustmentGroup { reason changes { name delta quantityAfterChange } }
userErrors { field message }
}
}"""
def gql(query, variables=None):
r = requests.post(
ENDPOINT,
json={"query": query, "variables": variables or {}},
headers={"X-Shopify-Access-Token": TOKEN, "Content-Type": "application/json"},
timeout=30,
)
r.raise_for_status()
body = r.json()
if body.get("errors"):
raise RuntimeError(body["errors"])
return body["data"]
def load_snapshot(path):
"""Read the trusted pre-import counts. Keyed by SKU, value is the correct
available quantity. This is the source of truth, not what Shopify has now.
"""
snapshot = {}
with open(path, newline="", encoding="utf-8") as fh:
for row in csv.DictReader(fh):
sku = (row.get("sku") or "").strip()
if not sku:
continue
snapshot[sku] = int(row["available"])
return snapshot
def plan_correction(sku, live_quantity, snapshot, max_adjust):
"""Pure decision: should this item be corrected, and to what?
Returns a dict with the delta and target quantity when a correction is
warranted, or None when the item should be left alone. An item is left
alone when it is missing from the snapshot (we have no trusted value to
restore), when it already matches, or when the drift is larger than
max_adjust, since a swing that big is more likely a second bad file than
real damage, and should go to a human instead of being auto-applied.
"""
if sku not in snapshot:
return None
target = snapshot[sku]
delta = target - live_quantity
if delta == 0:
return None
if abs(delta) > max_adjust:
return None
return {"sku": sku, "from": live_quantity, "to": target, "delta": delta}
def batches(items, size):
for i in range(0, len(items), size):
yield items[i:i + size]
def live_levels(location_id):
cursor = None
while True:
data = gql(LOCATION_LEVELS_QUERY, {"id": location_id, "cursor": cursor})["location"]
levels = data["inventoryLevels"]
for node in levels["nodes"]:
available = next(q["quantity"] for q in node["quantities"] if q["name"] == "available")
yield node["item"]["id"], node["item"]["sku"], available
if not levels["pageInfo"]["hasNextPage"]:
return
cursor = levels["pageInfo"]["endCursor"]
def apply_batch(location_id, corrections_by_item):
"""corrections_by_item: list of (inventory_item_id, live_quantity, target_quantity)."""
quantities = [
{
"inventoryItemId": item_id,
"locationId": location_id,
"quantity": target,
"compareQuantity": live_quantity,
}
for item_id, live_quantity, target in corrections_by_item
]
result = gql(
SET_QUANTITIES_MUTATION,
{
"input": {
"name": "available",
"reason": "correction",
"ignoreCompareQuantity": False,
"quantities": quantities,
}
},
)["inventorySetQuantities"]
if result["userErrors"]:
raise RuntimeError(result["userErrors"])
return result["inventoryAdjustmentGroup"]
def run():
snapshot = load_snapshot(SNAPSHOT_PATH)
pending = []
scanned = 0
for item_id, sku, live_quantity in live_levels(LOCATION_ID):
scanned += 1
decision = plan_correction(sku, live_quantity, snapshot, MAX_ADJUST)
if decision is None:
continue
log.info(
"SKU %s drifted: %d -> %d (delta %+d). %s",
sku, decision["from"], decision["to"], decision["delta"],
"would fix" if DRY_RUN else "fixing",
)
pending.append((item_id, live_quantity, decision["to"]))
fixed = 0
if not DRY_RUN:
for batch in batches(pending, BATCH_SIZE):
apply_batch(LOCATION_ID, batch)
fixed += len(batch)
time.sleep(0.5) # be gentle with the API between batches
else:
fixed = len(pending)
log.info(
"Done. Scanned %d item(s), %d %s.",
scanned, fixed, "to correct" if DRY_RUN else "corrected",
)
if __name__ == "__main__":
run()/**
* True up Shopify available inventory after a bad bulk import, safely.
*
* A bad import (a bad CSV, a stuck sync app) can stamp the wrong "available"
* quantity onto many items at once. This script reads a trusted snapshot (the
* counts you captured before the bad import, keyed by SKU and location), reads
* each item's live quantity from Shopify, and only corrects items where the
* drift is real and inside a sane guard. It writes with inventorySetQuantities
* using compareQuantity, so a sale that lands between the read and the write
* is never silently overwritten. Batched, paged, and safe to run again and
* again.
*/
import { readFileSync } from "node:fs";
import { pathToFileURL } from "node:url";
const SHOP = process.env.SHOPIFY_SHOP || "example.myshopify.com";
const TOKEN = process.env.SHOPIFY_ACCESS_TOKEN || "shpat_dummy";
const API_VERSION = process.env.SHOPIFY_API_VERSION || "2025-01";
const ENDPOINT = `https://${SHOP}/admin/api/${API_VERSION}/graphql.json`;
const LOCATION_ID = process.env.SHOPIFY_LOCATION_ID || "gid://shopify/Location/0";
const SNAPSHOT_PATH = process.env.SNAPSHOT_PATH || "snapshot.csv";
const MAX_ADJUST = Number(process.env.MAX_ADJUST || 500);
const BATCH_SIZE = Number(process.env.BATCH_SIZE || 25);
const DRY_RUN = (process.env.DRY_RUN || "true").toLowerCase() === "true";
/**
* Pure decision: should this item be corrected, and to what?
*
* Returns an object with the delta and target quantity when a correction is
* warranted, or null when the item should be left alone. An item is left
* alone when it is missing from the snapshot (no trusted value to restore),
* when it already matches, or when the drift is larger than maxAdjust, since
* a swing that big is more likely a second bad file than real damage, and
* should go to a human instead of being auto-applied.
*/
export function planCorrection(sku, liveQuantity, snapshot, maxAdjust) {
if (!(sku in snapshot)) return null;
const target = snapshot[sku];
const delta = target - liveQuantity;
if (delta === 0) return null;
if (Math.abs(delta) > maxAdjust) return null;
return { sku, from: liveQuantity, to: target, delta };
}
/** Read the trusted pre-import counts from a small CSV: sku,available */
export function loadSnapshot(text) {
const snapshot = {};
const lines = text.split(/\r?\n/).filter((l) => l.trim().length > 0);
const [header, ...rows] = lines;
const cols = header.split(",").map((c) => c.trim().toLowerCase());
const skuIdx = cols.indexOf("sku");
const availIdx = cols.indexOf("available");
for (const line of rows) {
const cells = line.split(",");
const sku = (cells[skuIdx] || "").trim();
if (!sku) continue;
snapshot[sku] = parseInt(cells[availIdx], 10);
}
return snapshot;
}
export function batches(items, size) {
const out = [];
for (let i = 0; i < items.length; i += size) out.push(items.slice(i, i + size));
return out;
}
async function gql(query, variables = {}) {
const res = await fetch(ENDPOINT, {
method: "POST",
headers: { "X-Shopify-Access-Token": TOKEN, "Content-Type": "application/json" },
body: JSON.stringify({ query, variables }),
});
if (!res.ok) throw new Error(`Shopify ${res.status}`);
const body = await res.json();
if (body.errors) throw new Error(JSON.stringify(body.errors));
return body.data;
}
const LOCATION_LEVELS_QUERY = `
query($id: ID!, $cursor: String) {
location(id: $id) {
inventoryLevels(first: 100, after: $cursor) {
pageInfo { hasNextPage endCursor }
nodes {
item { id sku }
quantities(names: ["available"]) { name quantity }
}
}
}
}`;
const SET_QUANTITIES_MUTATION = `
mutation($input: InventorySetQuantitiesInput!) {
inventorySetQuantities(input: $input) {
inventoryAdjustmentGroup { reason changes { name delta quantityAfterChange } }
userErrors { field message }
}
}`;
async function* liveLevels(locationId) {
let cursor = null;
while (true) {
const data = (await gql(LOCATION_LEVELS_QUERY, { id: locationId, cursor })).location;
const levels = data.inventoryLevels;
for (const node of levels.nodes) {
const available = node.quantities.find((q) => q.name === "available").quantity;
yield { itemId: node.item.id, sku: node.item.sku, available };
}
if (!levels.pageInfo.hasNextPage) return;
cursor = levels.pageInfo.endCursor;
}
}
async function applyBatch(locationId, corrections) {
const quantities = corrections.map(({ itemId, liveQuantity, target }) => ({
inventoryItemId: itemId,
locationId,
quantity: target,
compareQuantity: liveQuantity,
}));
const result = (
await gql(SET_QUANTITIES_MUTATION, {
input: { name: "available", reason: "correction", ignoreCompareQuantity: false, quantities },
})
).inventorySetQuantities;
if (result.userErrors.length) throw new Error(JSON.stringify(result.userErrors));
return result.inventoryAdjustmentGroup;
}
export async function run() {
const snapshot = loadSnapshot(readFileSync(SNAPSHOT_PATH, "utf-8"));
const pending = [];
let scanned = 0;
for await (const { itemId, sku, available } of liveLevels(LOCATION_ID)) {
scanned++;
const decision = planCorrection(sku, available, snapshot, MAX_ADJUST);
if (!decision) continue;
console.log(
`SKU ${sku} drifted: ${decision.from} -> ${decision.to} (delta ${decision.delta > 0 ? "+" : ""}${decision.delta}). ${DRY_RUN ? "would fix" : "fixing"}`,
);
pending.push({ itemId, liveQuantity: available, target: decision.to });
}
let fixed = 0;
if (!DRY_RUN) {
for (const batch of batches(pending, BATCH_SIZE)) {
await applyBatch(LOCATION_ID, batch);
fixed += batch.length;
await new Promise((r) => setTimeout(r, 500)); // be gentle with the API between batches
}
} else {
fixed = pending.length;
}
console.log(`Done. Scanned ${scanned} item(s), ${fixed} ${DRY_RUN ? "to correct" : "corrected"}.`);
}
if (import.meta.url === pathToFileURL(process.argv[1]).href) {
run().catch((err) => { console.error(err); process.exit(1); });
}Add a test
The decision rule is the part most worth testing, because it decides which items on your live store get overwritten. Because we kept plan_correction pure, the test needs no network, no CSV file, and no Shopify account. It just feeds in plain values and checks the answer.
from true_up_inventory import plan_correction, load_snapshot
import csv
import os
import tempfile
def test_no_correction_when_matching():
assert plan_correction("SKU-1", 10, {"SKU-1": 10}, max_adjust=500) is None
def test_no_correction_when_missing_from_snapshot():
assert plan_correction("SKU-1", 10, {}, max_adjust=500) is None
def test_correction_when_drift_positive():
decision = plan_correction("SKU-1", 3, {"SKU-1": 40}, max_adjust=500)
assert decision == {"sku": "SKU-1", "from": 3, "to": 40, "delta": 37}
def test_correction_when_drift_negative():
decision = plan_correction("SKU-1", 90, {"SKU-1": 12}, max_adjust=500)
assert decision == {"sku": "SKU-1", "from": 90, "to": 12, "delta": -78}
def test_no_correction_when_drift_exceeds_guard():
# a swing bigger than max_adjust looks like a second bad file, not real damage
assert plan_correction("SKU-1", 5, {"SKU-1": 5000}, max_adjust=500) is None
def test_correction_allowed_at_exact_guard_boundary():
decision = plan_correction("SKU-1", 0, {"SKU-1": 500}, max_adjust=500)
assert decision["delta"] == 500
def test_load_snapshot_reads_csv_by_sku():
with tempfile.TemporaryDirectory() as tmp:
path = os.path.join(tmp, "snapshot.csv")
with open(path, "w", newline="", encoding="utf-8") as fh:
writer = csv.writer(fh)
writer.writerow(["sku", "available"])
writer.writerow(["SKU-1", "40"])
writer.writerow(["SKU-2", "0"])
writer.writerow(["", "99"]) # blank sku is skipped
snapshot = load_snapshot(path)
assert snapshot == {"SKU-1": 40, "SKU-2": 0}import { test } from "node:test";
import assert from "node:assert/strict";
import { planCorrection, loadSnapshot, batches } from "./true-up-inventory.js";
test("no correction when matching", () => {
assert.equal(planCorrection("SKU-1", 10, { "SKU-1": 10 }, 500), null);
});
test("no correction when missing from snapshot", () => {
assert.equal(planCorrection("SKU-1", 10, {}, 500), null);
});
test("correction when drift positive", () => {
assert.deepEqual(
planCorrection("SKU-1", 3, { "SKU-1": 40 }, 500),
{ sku: "SKU-1", from: 3, to: 40, delta: 37 },
);
});
test("correction when drift negative", () => {
assert.deepEqual(
planCorrection("SKU-1", 90, { "SKU-1": 12 }, 500),
{ sku: "SKU-1", from: 90, to: 12, delta: -78 },
);
});
test("no correction when drift exceeds guard", () => {
assert.equal(planCorrection("SKU-1", 5, { "SKU-1": 5000 }, 500), null);
});
test("correction allowed at exact guard boundary", () => {
const decision = planCorrection("SKU-1", 0, { "SKU-1": 500 }, 500);
assert.equal(decision.delta, 500);
});
test("loadSnapshot reads csv by sku", () => {
const csv = "sku,available\nSKU-1,40\nSKU-2,0\n,99\n";
assert.deepEqual(loadSnapshot(csv), { "SKU-1": 40, "SKU-2": 0 });
});
test("batches splits into chunks of the given size", () => {
assert.deepEqual(batches([1, 2, 3, 4, 5], 2), [[1, 2], [3, 4], [5]]);
});
test("batches returns empty array for empty input", () => {
assert.deepEqual(batches([], 25), []);
});Case studies
A weight column landed in the quantity field
A homeware brand ran a bulk update from a spreadsheet a warehouse partner sent over. The partner had reordered the columns, so what the import tool read as "available" was actually the item's weight in grams. Overnight, dozens of items showed thousands of units in stock, and a handful showed zero.
The team pulled the export they had taken the day before, saved it as a snapshot, and ran the script in dry run. It listed exactly which SKUs had drifted and by how much, they agreed the list looked right, and the real run corrected only those SKUs, batch by batch, without touching anything that import had not actually broken.
Last month's numbers came back from the dead
A seasonal store kept monthly stock exports as backups. Someone grabbed the wrong file from a shared folder and ran it as a bulk update, quietly rolling every item back to counts from four weeks earlier, right before a big restock and a busy sale weekend.
Because they had a snapshot from the morning of the bad import, the true-up caught every SKU that had drifted, skipped the handful that coincidentally matched the stale file, and applied the rest in small batches with compare-and-set writes, so nothing was lost even though sales kept coming in during the run.
After this runs, the store matches the snapshot you trusted, and nothing beyond the real drift was touched. Sales that happened during the run were respected because of the compare-and-set write, and any swing too large to trust automatically got left for a human to check by hand. Keep the snapshot habit going, since the next bad import will need one too.
FAQ
Why did a bulk import break my inventory counts?
A bulk import usually writes an absolute available quantity for every row in the file. If the file had the wrong column mapped, was exported from a stale system, or was uploaded twice, every item it touched now carries that wrong number, and Shopify has no way to know the file itself was bad.
Is it safe to bulk correct inventory with a script?
Yes, when the script compares each item against a trusted snapshot taken before the bad import, skips anything the snapshot does not cover, skips any drift bigger than a sane guard, writes with compareQuantity so a sale that happens in between is never overwritten, and runs in dry run first.
What does compareQuantity do on inventorySetQuantities?
compareQuantity makes the write a compare-and-set. Shopify only applies the new quantity if the value it currently has still matches what you read a moment ago. If a sale changed the count in between, the write is rejected instead of silently overwriting a number you never saw.
Related field notes
Citations
On the problem:
- Shopify Help Center: updating inventory quantities in bulk with a CSV file. help.shopify.com/en/manual/products/inventory/getting-started/inventory-quantities-csv
- Shopify Help Center: understanding available, on hand, and committed inventory. help.shopify.com/en/manual/products/inventory/inventory-quantities
- Shopify Community: bulk inventory update overwrote correct stock counts. community.shopify.com shopify apis and sdks
On the solution:
- Shopify Admin GraphQL: the
inventorySetQuantitiesmutation and its compare-and-set behavior. shopify.dev/docs/api/admin-graphql/latest/mutations/inventorySetQuantities - Shopify Admin GraphQL: the
InventoryLevelobject and itsquantitiesfield. shopify.dev/docs/api/admin-graphql/latest/objects/InventoryLevel - Shopify Admin GraphQL: the
Locationobject and itsinventoryLevelsconnection. shopify.dev/docs/api/admin-graphql/latest/objects/Location
Stuck on a tricky one?
If you have a problem in Shopify orders, payments, subscriptions, inventory, or fulfillment that you would rather hand off, this is the kind of work I do. Message me and we can work through it together.
Did this untangle your catalog?
If this saved you from a pile of wrong counts or a scary re-import, you can buy me a coffee. It is the best way to keep these field notes free and growing.
Buy me a coffee on Ko-fi