Repair Catalog Import
CSV import ignores the variant inventory quantity column
You upload a product CSV with a proper Variant Inventory Quantity column, the import finishes with a green checkmark, and every product shows up in the catalog looking exactly right. Then the storefront says everything is out of stock. Nothing errored. Nothing warned you. The quantity column was simply never read. Here is why Medusa v2's import leaves every variant with no usable stock, and a small script that finds every variant this happened to and repairs it from the same CSV.
In Medusa v2, stock is no longer a field on the variant itself. It lives on a location_levels record under a linked inventory_item, in the stocked_quantity field. importProductsWorkflow is supposed to create that inventory item for each variant and then create a location level for it from the CSV quantity, but its CSV parsing and normalization step does not map the legacy Variant Inventory Quantity column to that location level creation step. This is tracked upstream as medusajs/medusa#11605 and #9357, a real gap between the old v1 CSV template semantics and the v2 inventory item and location level data model. Every imported variant ends up with an inventory item that has no, or zero, stocked quantity, no matter what the source CSV said. Run a small Python or Node.js script that compares each imported variant's actual location levels against the source CSV row for its SKU, and reports or repairs every mismatch. Full code, tests, and a dry run guard are below.
The problem in plain words
In Medusa v1, a variant carried its own inventory quantity, so a CSV column called Variant Inventory Quantity mapped directly onto that field. Medusa v2 split inventory out into its own module. A variant is now linked to an inventory_item, and the actual count on hand lives one level further down, on a location_levels record scoped to a specific stock location, in the stocked_quantity field.
The import workflow still accepts the same familiar CSV template, including the Variant Inventory Quantity column, so the upload screen looks unchanged and nothing about the file format complains. But importProductsWorkflow's CSV normalization step, the part of the code that turns each parsed row into workflow input, was written against the newer data model and never wires that column's value into a location level creation step. The workflow does create an inventory item for each variant, since a variant needs one to be purchasable at all, but it either skips creating a location level entirely or creates one with no quantity set. The CSV said 200 units. Medusa recorded zero, or nothing.
Why it happens
Since a v2 variant's sellable quantity comes from a location_levels record and not from a field on the variant, and the CSV normalization code was never updated to bridge the old column to that new step, the failure shows up on every import that relies on the quantity column instead of setting stock by hand afterward. A few common ways stores end up here:
- A store migrating from Shopify, WooCommerce, or Medusa v1 exports a CSV with the familiar
Variant Inventory Quantitycolumn and expects the import to seed opening stock the same way the old template did. - The import finishes with a success message and every product, price, and option looks correct, so nobody thinks to check inventory until the storefront reports the item cannot be added to the cart.
- A merchandiser re-imports the same CSV later to update prices, and it quietly overwrites nothing on the inventory side, since the quantity column was never touched by the workflow in the first place.
- A store manages inventory across multiple stock locations, and even a fix that creates one location level still leaves every other location at zero, since the CSV template only ever had one quantity column.
This is a common source of confusion because the import UI gives no signal that anything was skipped. There is no failed row, no validation warning, nothing in the response body pointing at inventory. The mismatch only surfaces downstream, once a customer or a QA pass notices every new product is unbuyable. See the citations at the end for the exact issues and the docs on how v2 inventory actually works.
A missing stocked quantity after import is not a data corruption problem, it is a mapping gap. The inventory item usually does exist, it is just missing the one write that would give it a real count at a real stock location. So the safe pattern is not "set every imported variant's stock to whatever looks reasonable." It is "compare each variant's actual location levels against the CSV row that was supposed to seed it, and only touch the ones that are still at zero when the source clearly said otherwise."
The fix, as a flow
We do not change the import step. We add a check that runs after an import batch, reads back the variants Medusa actually created, and compares each one's inventory location levels against the original CSV. Rows the import got right are left alone. Rows where the location level is missing or stuck at zero, despite a CSV quantity above zero, get flagged, and only get written when a human turns off dry run.
Build it step by step
Get an admin token
Exchange an admin email and password for a JWT at POST /auth/user/emailpass, then send it as Authorization: Bearer <token> on every /admin/* call. Keep the backend URL and credentials in environment variables, never in the file.
pip install requests
export MEDUSA_BACKEND_URL="http://localhost:9000"
export MEDUSA_ADMIN_EMAIL="admin@example.com"
export MEDUSA_ADMIN_PASSWORD="supersecret"
export DRY_RUN="true" # start safe, change to false to write// Node 18+ has fetch built in, no dependencies needed
export MEDUSA_BACKEND_URL="http://localhost:9000"
export MEDUSA_ADMIN_EMAIL="admin@example.com"
export MEDUSA_ADMIN_PASSWORD="supersecret"
export DRY_RUN="true" // start safe, change to false to writeList the variants an import batch just created
Ask for products with fields=id,title,*variants,*variants.inventory_items,*variants.inventory_items.inventory, filtered by a metadata tag you set during import or by a created_at range, so you are only checking the batch you care about. Each variant's inventory_items array gives you the inventory_item_id to check next.
import os, requests
BACKEND_URL = os.environ["MEDUSA_BACKEND_URL"]
ADMIN_EMAIL = os.environ["MEDUSA_ADMIN_EMAIL"]
ADMIN_PASSWORD = os.environ["MEDUSA_ADMIN_PASSWORD"]
def get_admin_token():
r = requests.post(
f"{BACKEND_URL}/auth/user/emailpass",
json={"email": ADMIN_EMAIL, "password": ADMIN_PASSWORD},
timeout=30,
)
r.raise_for_status()
return r.json()["token"]
def get_imported_products(token, batch_tag):
r = requests.get(
f"{BACKEND_URL}/admin/products",
headers={"Authorization": f"Bearer {token}"},
params={
"q": batch_tag,
"fields": "id,title,*variants,*variants.inventory_items,*variants.inventory_items.inventory",
},
timeout=30,
)
r.raise_for_status()
return r.json()["products"]import Medusa from "@medusajs/js-sdk";
const sdk = new Medusa({
baseUrl: process.env.MEDUSA_BACKEND_URL,
auth: { type: "jwt" },
});
async function login() {
return sdk.auth.login("user", "emailpass", {
email: process.env.MEDUSA_ADMIN_EMAIL,
password: process.env.MEDUSA_ADMIN_PASSWORD,
});
}
async function getImportedProducts(batchTag) {
const { products } = await sdk.admin.product.list({
q: batchTag,
fields: "id,title,*variants,*variants.inventory_items,*variants.inventory_items.inventory",
});
return products;
}Read the actual location levels for each variant
For each variant's inventory_item_id, call GET /admin/inventory-items/{inventory_item_id}/location-levels to get the real stocked_quantity at each stock location. If the array is empty, or every level reads zero, the import left this variant with no usable stock.
def get_location_levels(token, inventory_item_id):
r = requests.get(
f"{BACKEND_URL}/admin/inventory-items/{inventory_item_id}/location-levels",
headers={"Authorization": f"Bearer {token}"},
params={"fields": "location_id,stocked_quantity"},
timeout=30,
)
r.raise_for_status()
return r.json()["inventory_levels"]async function getLocationLevels(inventoryItemId) {
const { inventory_levels } = await sdk.admin.inventoryItem.listLocationLevels(inventoryItemId, {
fields: "location_id,stocked_quantity",
});
return inventory_levels;
}Read the quantity the CSV actually asked for
Parse the same source CSV, keyed by SKU, so you have the original Variant Inventory Quantity value the import was supposed to use. This is the ground truth the decision function compares against, since Medusa's own data cannot tell you what the CSV originally intended.
import csv
def read_csv_rows_by_sku(csv_path):
rows = {}
with open(csv_path, newline="", encoding="utf-8") as f:
for row in csv.DictReader(f):
sku = row.get("Variant SKU") or row.get("sku")
qty = row.get("Variant Inventory Quantity") or row.get("variant_inventory_quantity") or "0"
if sku:
rows[sku] = {"sku": sku, "variantInventoryQuantity": int(float(qty))}
return rowsimport { readFileSync } from "node:fs";
export function readCsvRowsBySku(csvPath) {
const text = readFileSync(csvPath, "utf-8");
const [headerLine, ...lines] = text.trim().split("\n");
const headers = headerLine.split(",").map((h) => h.trim());
const skuIdx = headers.indexOf("Variant SKU");
const qtyIdx = headers.indexOf("Variant Inventory Quantity");
const rows = {};
for (const line of lines) {
const cols = line.split(",");
const sku = cols[skuIdx];
const qty = Number(cols[qtyIdx] || "0");
if (sku) rows[sku] = { sku, variantInventoryQuantity: qty };
}
return rows;
}Decide, with one pure function
Keep the decision in its own function that takes the CSV row, the variant, the variant's actual location levels, and a default stock location id, and returns the repair action to take, or nothing. It is strict on purpose. If the CSV never expected stock, or the variant has no inventory item at all, it does nothing, since a script cannot safely invent an inventory item. No I/O, so it is directly testable with fixtures.
def decide_inventory_repair(csv_row, variant, location_levels, default_location_id):
if csv_row.get("variantInventoryQuantity", 0) <= 0:
return None
if not variant.get("inventoryItemId"):
return None
if not location_levels:
return {
"action": "create_level",
"inventoryItemId": variant["inventoryItemId"],
"locationId": default_location_id,
"fromQty": 0,
"toQty": csv_row["variantInventoryQuantity"],
}
level = next((lvl for lvl in location_levels if lvl.get("location_id") == default_location_id), None)
if level is None:
return {
"action": "create_level",
"inventoryItemId": variant["inventoryItemId"],
"locationId": default_location_id,
"fromQty": 0,
"toQty": csv_row["variantInventoryQuantity"],
}
if level.get("stocked_quantity", 0) != csv_row["variantInventoryQuantity"]:
return {
"action": "update_level",
"inventoryItemId": variant["inventoryItemId"],
"locationId": default_location_id,
"fromQty": level.get("stocked_quantity", 0),
"toQty": csv_row["variantInventoryQuantity"],
}
return Noneexport function decideInventoryRepair(csvRow, variant, locationLevels, defaultLocationId) {
if ((csvRow.variantInventoryQuantity || 0) <= 0) return null;
if (!variant.inventoryItemId) return null;
if (!locationLevels.length) {
return {
action: "create_level",
inventoryItemId: variant.inventoryItemId,
locationId: defaultLocationId,
fromQty: 0,
toQty: csvRow.variantInventoryQuantity,
};
}
const level = locationLevels.find((lvl) => lvl.location_id === defaultLocationId) || null;
if (!level) {
return {
action: "create_level",
inventoryItemId: variant.inventoryItemId,
locationId: defaultLocationId,
fromQty: 0,
toQty: csvRow.variantInventoryQuantity,
};
}
if ((level.stocked_quantity || 0) !== csvRow.variantInventoryQuantity) {
return {
action: "update_level",
inventoryItemId: variant.inventoryItemId,
locationId: defaultLocationId,
fromQty: level.stocked_quantity || 0,
toQty: csvRow.variantInventoryQuantity,
};
}
return null;
}Wire it together with a dry run guard
The run loop walks each imported variant, resolves the repair action, and logs the intended {inventoryItemId, locationId, fromQty, toQty} triple. When a location level is missing, it calls POST /admin/inventory-items/{inventory_item_id}/location-levels with the CSV quantity. When a level exists but is stuck at zero, it calls POST /admin/inventory-items/{inventory_item_id}/location-levels/{location_id} to update it. Both writes only happen when DRY_RUN is off.
Always start with DRY_RUN=true. Picking the right stock location and reconciling any reservations already in flight is store specific, so confirm the default location id yourself before writing, and never let the script guess a location it was not told about.
The full code
Here is the complete script in one file for each language. It reads settings from the environment, logs every repair it finds, and only writes a location level create or update when dry run is off.
"""Repair Medusa variants whose CSV import never set a stocked quantity.
In Medusa v2, stock lives on a location_levels record under a linked
inventory_item, in stocked_quantity, not on the variant itself.
importProductsWorkflow creates the inventory item for each variant, but its
CSV normalization step does not map the legacy Variant Inventory Quantity
column to a location level creation step, tracked upstream as
medusajs/medusa issues 11605 and 9357. Every imported variant can end up
with no location level, or one stuck at zero, no matter what the source CSV
said. This reads back the variants an import batch created, compares each
one's actual location levels against the source CSV row for its SKU, and
either logs or writes the missing stocked_quantity. Run once after an
import. Safe to run again and again.
Guide: https://www.allanninal.dev/medusa/csv-import-ignores-inventory-quantity/
"""
import os
import csv
import logging
import requests
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("repair_import_inventory")
BACKEND_URL = os.environ.get("MEDUSA_BACKEND_URL", "http://localhost:9000")
ADMIN_EMAIL = os.environ.get("MEDUSA_ADMIN_EMAIL", "admin@example.com")
ADMIN_PASSWORD = os.environ.get("MEDUSA_ADMIN_PASSWORD", "supersecret")
DRY_RUN = os.environ.get("DRY_RUN", "true").lower() == "true"
BATCH_TAG = os.environ.get("IMPORT_BATCH_TAG", "")
CSV_PATH = os.environ.get("IMPORT_CSV_PATH", "import.csv")
DEFAULT_LOCATION_ID = os.environ.get("DEFAULT_LOCATION_ID", "")
def get_admin_token():
r = requests.post(
f"{BACKEND_URL}/auth/user/emailpass",
json={"email": ADMIN_EMAIL, "password": ADMIN_PASSWORD},
timeout=30,
)
r.raise_for_status()
return r.json()["token"]
def get_imported_products(token, batch_tag):
r = requests.get(
f"{BACKEND_URL}/admin/products",
headers={"Authorization": f"Bearer {token}"},
params={
"q": batch_tag,
"fields": "id,title,*variants,*variants.inventory_items,*variants.inventory_items.inventory",
},
timeout=30,
)
r.raise_for_status()
return r.json()["products"]
def get_location_levels(token, inventory_item_id):
r = requests.get(
f"{BACKEND_URL}/admin/inventory-items/{inventory_item_id}/location-levels",
headers={"Authorization": f"Bearer {token}"},
params={"fields": "location_id,stocked_quantity"},
timeout=30,
)
r.raise_for_status()
return r.json()["inventory_levels"]
def read_csv_rows_by_sku(csv_path):
rows = {}
with open(csv_path, newline="", encoding="utf-8") as f:
for row in csv.DictReader(f):
sku = row.get("Variant SKU") or row.get("sku")
qty = row.get("Variant Inventory Quantity") or row.get("variant_inventory_quantity") or "0"
if sku:
rows[sku] = {"sku": sku, "variantInventoryQuantity": int(float(qty))}
return rows
def decide_inventory_repair(csv_row, variant, location_levels, default_location_id):
"""Pure decision function. No I/O.
csv_row: {"sku": str, "variantInventoryQuantity": int}
variant: {"id": str, "sku": str, "inventoryItemId": str | None}
location_levels: [{"location_id": str, "stocked_quantity": number}, ...]
default_location_id: str
Returns a repair action dict, or None if nothing needs to change.
"""
if csv_row.get("variantInventoryQuantity", 0) <= 0:
return None
if not variant.get("inventoryItemId"):
return None
if not location_levels:
return {
"action": "create_level",
"inventoryItemId": variant["inventoryItemId"],
"locationId": default_location_id,
"fromQty": 0,
"toQty": csv_row["variantInventoryQuantity"],
}
level = next((lvl for lvl in location_levels if lvl.get("location_id") == default_location_id), None)
if level is None:
return {
"action": "create_level",
"inventoryItemId": variant["inventoryItemId"],
"locationId": default_location_id,
"fromQty": 0,
"toQty": csv_row["variantInventoryQuantity"],
}
if level.get("stocked_quantity", 0) != csv_row["variantInventoryQuantity"]:
return {
"action": "update_level",
"inventoryItemId": variant["inventoryItemId"],
"locationId": default_location_id,
"fromQty": level.get("stocked_quantity", 0),
"toQty": csv_row["variantInventoryQuantity"],
}
return None
def create_location_level(token, inventory_item_id, location_id, stocked_quantity):
r = requests.post(
f"{BACKEND_URL}/admin/inventory-items/{inventory_item_id}/location-levels",
headers={"Authorization": f"Bearer {token}"},
json={"location_id": location_id, "stocked_quantity": stocked_quantity},
timeout=30,
)
r.raise_for_status()
return r.json()
def update_location_level(token, inventory_item_id, location_id, stocked_quantity):
r = requests.post(
f"{BACKEND_URL}/admin/inventory-items/{inventory_item_id}/location-levels/{location_id}",
headers={"Authorization": f"Bearer {token}"},
json={"stocked_quantity": stocked_quantity},
timeout=30,
)
r.raise_for_status()
return r.json()
def run():
if not DEFAULT_LOCATION_ID:
raise SystemExit("Set DEFAULT_LOCATION_ID to the stock location the CSV quantity should land on.")
token = get_admin_token()
csv_rows = read_csv_rows_by_sku(CSV_PATH)
products = get_imported_products(token, BATCH_TAG)
repaired = 0
skipped_no_inventory_item = 0
for product in products:
for variant in product.get("variants") or []:
sku = variant.get("sku")
csv_row = csv_rows.get(sku)
if not csv_row:
continue
inventory_items = variant.get("inventory_items") or []
inventory_item_id = None
if inventory_items:
inventory_item_id = (inventory_items[0].get("inventory") or {}).get("id") or inventory_items[0].get("inventory_item_id")
variant_input = {
"id": variant["id"],
"sku": sku,
"inventoryItemId": inventory_item_id,
}
if csv_row["variantInventoryQuantity"] > 0 and not inventory_item_id:
skipped_no_inventory_item += 1
log.warning(
"Variant %s (SKU %s): CSV expected %s units but has no inventory item, flagging for manual review",
variant["id"], sku, csv_row["variantInventoryQuantity"],
)
continue
location_levels = get_location_levels(token, inventory_item_id) if inventory_item_id else []
decision = decide_inventory_repair(csv_row, variant_input, location_levels, DEFAULT_LOCATION_ID)
if decision is None:
continue
log.info(
"%s variant %s (SKU %s): location %s, %s -> %s",
"Would repair" if DRY_RUN else "Repairing",
variant["id"], sku, decision["locationId"], decision["fromQty"], decision["toQty"],
)
if not DRY_RUN:
if decision["action"] == "create_level":
create_location_level(token, decision["inventoryItemId"], decision["locationId"], decision["toQty"])
elif decision["action"] == "update_level":
update_location_level(token, decision["inventoryItemId"], decision["locationId"], decision["toQty"])
repaired += 1
log.info("Done. %d variant(s) %s, %d flagged with no inventory item.",
repaired, "to repair" if DRY_RUN else "repaired", skipped_no_inventory_item)
if __name__ == "__main__":
run()/**
* Repair Medusa variants whose CSV import never set a stocked quantity.
*
* In Medusa v2, stock lives on a location_levels record under a linked
* inventory_item, in stocked_quantity, not on the variant itself.
* importProductsWorkflow creates the inventory item for each variant, but
* its CSV normalization step does not map the legacy Variant Inventory
* Quantity column to a location level creation step, tracked upstream as
* medusajs/medusa issues 11605 and 9357. Every imported variant can end up
* with no location level, or one stuck at zero, no matter what the source
* CSV said. This reads back the variants an import batch created, compares
* each one's actual location levels against the source CSV row for its
* SKU, and either logs or writes the missing stocked_quantity. Run once
* after an import.
*
* Guide: https://www.allanninal.dev/medusa/csv-import-ignores-inventory-quantity/
*/
import { pathToFileURL } from "node:url";
import { readFileSync } from "node:fs";
const BACKEND_URL = process.env.MEDUSA_BACKEND_URL || "http://localhost:9000";
const ADMIN_EMAIL = process.env.MEDUSA_ADMIN_EMAIL || "admin@example.com";
const ADMIN_PASSWORD = process.env.MEDUSA_ADMIN_PASSWORD || "supersecret";
const DRY_RUN = (process.env.DRY_RUN || "true").toLowerCase() === "true";
const BATCH_TAG = process.env.IMPORT_BATCH_TAG || "";
const CSV_PATH = process.env.IMPORT_CSV_PATH || "import.csv";
const DEFAULT_LOCATION_ID = process.env.DEFAULT_LOCATION_ID || "";
export function readCsvRowsBySku(csvPath) {
const text = readFileSync(csvPath, "utf-8");
const [headerLine, ...lines] = text.trim().split("\n");
const headers = headerLine.split(",").map((h) => h.trim());
const skuIdx = headers.indexOf("Variant SKU");
const qtyIdx = headers.indexOf("Variant Inventory Quantity");
const rows = {};
for (const line of lines) {
const cols = line.split(",");
const sku = cols[skuIdx];
const qty = Number(cols[qtyIdx] || "0");
if (sku) rows[sku] = { sku, variantInventoryQuantity: qty };
}
return rows;
}
export function decideInventoryRepair(csvRow, variant, locationLevels, defaultLocationId) {
if ((csvRow.variantInventoryQuantity || 0) <= 0) return null;
if (!variant.inventoryItemId) return null;
if (!locationLevels.length) {
return {
action: "create_level",
inventoryItemId: variant.inventoryItemId,
locationId: defaultLocationId,
fromQty: 0,
toQty: csvRow.variantInventoryQuantity,
};
}
const level = locationLevels.find((lvl) => lvl.location_id === defaultLocationId) || null;
if (!level) {
return {
action: "create_level",
inventoryItemId: variant.inventoryItemId,
locationId: defaultLocationId,
fromQty: 0,
toQty: csvRow.variantInventoryQuantity,
};
}
if ((level.stocked_quantity || 0) !== csvRow.variantInventoryQuantity) {
return {
action: "update_level",
inventoryItemId: variant.inventoryItemId,
locationId: defaultLocationId,
fromQty: level.stocked_quantity || 0,
toQty: csvRow.variantInventoryQuantity,
};
}
return null;
}
async function getSdk() {
const { default: Medusa } = await import("@medusajs/js-sdk");
const sdk = new Medusa({ baseUrl: BACKEND_URL, auth: { type: "jwt" } });
await sdk.auth.login("user", "emailpass", { email: ADMIN_EMAIL, password: ADMIN_PASSWORD });
return sdk;
}
async function getImportedProducts(sdk, batchTag) {
const { products } = await sdk.admin.product.list({
q: batchTag,
fields: "id,title,*variants,*variants.inventory_items,*variants.inventory_items.inventory",
});
return products;
}
async function getLocationLevels(sdk, inventoryItemId) {
const { inventory_levels } = await sdk.admin.inventoryItem.listLocationLevels(inventoryItemId, {
fields: "location_id,stocked_quantity",
});
return inventory_levels;
}
async function createLocationLevel(sdk, inventoryItemId, locationId, stockedQuantity) {
return sdk.admin.inventoryItem.createLocationLevel(inventoryItemId, {
location_id: locationId,
stocked_quantity: stockedQuantity,
});
}
async function updateLocationLevel(sdk, inventoryItemId, locationId, stockedQuantity) {
return sdk.admin.inventoryItem.updateLocationLevel(inventoryItemId, locationId, {
stocked_quantity: stockedQuantity,
});
}
export async function run() {
if (!DEFAULT_LOCATION_ID) {
throw new Error("Set DEFAULT_LOCATION_ID to the stock location the CSV quantity should land on.");
}
const sdk = await getSdk();
const csvRows = readCsvRowsBySku(CSV_PATH);
const products = await getImportedProducts(sdk, BATCH_TAG);
let repaired = 0;
let skippedNoInventoryItem = 0;
for (const product of products) {
for (const variant of product.variants || []) {
const sku = variant.sku;
const csvRow = csvRows[sku];
if (!csvRow) continue;
const inventoryItems = variant.inventory_items || [];
const inventoryItemId = inventoryItems[0]?.inventory?.id || inventoryItems[0]?.inventory_item_id || null;
const variantInput = { id: variant.id, sku, inventoryItemId };
if (csvRow.variantInventoryQuantity > 0 && !inventoryItemId) {
skippedNoInventoryItem++;
console.warn(
`Variant ${variant.id} (SKU ${sku}): CSV expected ${csvRow.variantInventoryQuantity} units but has no inventory item, flagging for manual review`
);
continue;
}
const locationLevels = inventoryItemId ? await getLocationLevels(sdk, inventoryItemId) : [];
const decision = decideInventoryRepair(csvRow, variantInput, locationLevels, DEFAULT_LOCATION_ID);
if (!decision) continue;
console.log(
`${DRY_RUN ? "Would repair" : "Repairing"} variant ${variant.id} (SKU ${sku}): location ${decision.locationId}, ${decision.fromQty} -> ${decision.toQty}`
);
if (!DRY_RUN) {
if (decision.action === "create_level") {
await createLocationLevel(sdk, decision.inventoryItemId, decision.locationId, decision.toQty);
} else if (decision.action === "update_level") {
await updateLocationLevel(sdk, decision.inventoryItemId, decision.locationId, decision.toQty);
}
}
repaired++;
}
}
console.log(
`Done. ${repaired} variant(s) ${DRY_RUN ? "to repair" : "repaired"}, ${skippedNoInventoryItem} flagged with no inventory item.`
);
}
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 whether a real variant gets its stock rewritten. Because we kept decide_inventory_repair pure, the test needs no network and no Medusa backend. It just feeds in plain objects and checks the answer.
from repair_import_inventory import decide_inventory_repair
def csv_row(qty=200, sku="SKU-1"):
return {"sku": sku, "variantInventoryQuantity": qty}
def variant(inventory_item_id="iitem_1", sku="SKU-1"):
return {"id": "variant_1", "sku": sku, "inventoryItemId": inventory_item_id}
def level(location_id="sloc_default", stocked_quantity=0):
return {"location_id": location_id, "stocked_quantity": stocked_quantity}
def test_csv_had_no_quantity_does_nothing():
assert decide_inventory_repair(csv_row(qty=0), variant(), [], "sloc_default") is None
def test_no_inventory_item_does_nothing():
result = decide_inventory_repair(csv_row(), variant(inventory_item_id=None), [], "sloc_default")
assert result is None
def test_empty_location_levels_creates_a_level():
result = decide_inventory_repair(csv_row(), variant(), [], "sloc_default")
assert result == {
"action": "create_level",
"inventoryItemId": "iitem_1",
"locationId": "sloc_default",
"fromQty": 0,
"toQty": 200,
}
def test_level_at_zero_updates_to_csv_quantity():
result = decide_inventory_repair(csv_row(), variant(), [level(stocked_quantity=0)], "sloc_default")
assert result == {
"action": "update_level",
"inventoryItemId": "iitem_1",
"locationId": "sloc_default",
"fromQty": 0,
"toQty": 200,
}
def test_level_already_matching_csv_does_nothing():
result = decide_inventory_repair(csv_row(), variant(), [level(stocked_quantity=200)], "sloc_default")
assert result is None
def test_level_at_a_different_location_creates_the_default_level():
result = decide_inventory_repair(csv_row(), variant(), [level(location_id="sloc_other", stocked_quantity=50)], "sloc_default")
assert result["action"] == "create_level"
assert result["toQty"] == 200import { test } from "node:test";
import assert from "node:assert/strict";
import { decideInventoryRepair } from "./repair-import-inventory.js";
const csvRow = (qty = 200, sku = "SKU-1") => ({ sku, variantInventoryQuantity: qty });
const variant = (inventoryItemId = "iitem_1", sku = "SKU-1") => ({ id: "variant_1", sku, inventoryItemId });
const level = (locationId = "sloc_default", stockedQuantity = 0) => ({ location_id: locationId, stocked_quantity: stockedQuantity });
test("CSV had no quantity does nothing", () => {
assert.equal(decideInventoryRepair(csvRow(0), variant(), [], "sloc_default"), null);
});
test("no inventory item does nothing", () => {
const result = decideInventoryRepair(csvRow(), variant(null), [], "sloc_default");
assert.equal(result, null);
});
test("empty location levels creates a level", () => {
const result = decideInventoryRepair(csvRow(), variant(), [], "sloc_default");
assert.deepEqual(result, {
action: "create_level",
inventoryItemId: "iitem_1",
locationId: "sloc_default",
fromQty: 0,
toQty: 200,
});
});
test("level at zero updates to CSV quantity", () => {
const result = decideInventoryRepair(csvRow(), variant(), [level("sloc_default", 0)], "sloc_default");
assert.deepEqual(result, {
action: "update_level",
inventoryItemId: "iitem_1",
locationId: "sloc_default",
fromQty: 0,
toQty: 200,
});
});
test("level already matching CSV does nothing", () => {
const result = decideInventoryRepair(csvRow(), variant(), [level("sloc_default", 200)], "sloc_default");
assert.equal(result, null);
});
test("level at a different location creates the default level", () => {
const result = decideInventoryRepair(csvRow(), variant(), [level("sloc_other", 50)], "sloc_default");
assert.equal(result.action, "create_level");
assert.equal(result.toQty, 200);
});Case studies
Two thousand variants, zero stock, one launch day
A home goods brand migrated from Shopify to Medusa v2 and exported the full catalog to CSV, quantity column included, expecting the import to seed opening stock the same way the old platform's importer had. The import ran clean, every product and price looked right in the admin, and the store launched on schedule.
Within an hour, support tickets started asking why nothing could be added to the cart. Running the repair script against the launch batch and the original CSV found that all two thousand variants had inventory items with no location level at all. A dry run confirmed the exact list, and turning it off filled in the correct opening stock at the default warehouse in one pass.
A weekly re-import that never touched inventory
A distributor re-uploaded a supplier CSV every week to refresh prices and descriptions, assuming the same file would also keep quantities current the way it once had on their old store. New products from the feed appeared correctly priced but permanently unbuyable, since the import created their inventory items with a location level stuck at zero.
Once the team wired the repair script into the same job that ran the weekly import, it started catching every new SKU whose quantity never made it through, before the next sales cycle began, instead of finding out from a customer.
After this runs against an import batch, every variant whose CSV row expected real stock either already has it, or gets a clearly logged repair with the exact inventory item, stock location, and quantity change. Nothing gets touched that was already correct, and nothing gets guessed at a location nobody confirmed. The catalog finishes an import actually sellable, not just visually complete.
FAQ
Why does my Medusa CSV import leave every variant with zero stock?
In Medusa v2, stock is not a field on the variant. It lives on a location level under a linked inventory item, in the stocked_quantity field. The v2 CSV parsing step used by importProductsWorkflow does not map the legacy Variant Inventory Quantity column to that location level creation step, so every imported variant gets an inventory item with no location level, or a location level stuck at zero, no matter what quantity the CSV row actually had.
Is it safe to automatically write the CSV quantity into Medusa after import?
It is safe once you scope it to variants whose location levels are actually missing or at zero while their source CSV row had a quantity above zero, and you always confirm the target stock location yourself. The default in the script below is DRY_RUN true, which only logs the intended change, because picking the right stock location and not clobbering a legitimate zero is store specific.
How do I check whether a Medusa variant actually has stock after an import?
Fetch the product with fields=id,title,*variants,*variants.inventory_items,*variants.inventory_items.inventory to get each variant's inventory_item_id. Then call GET /admin/inventory-items/{inventory_item_id}/location-levels for each one. If every location level shows stocked_quantity 0, or the list is empty, the variant has no usable stock regardless of what the import screen reported.
Related field notes
Citations
On the problem:
- GitHub Issue: Not setting Variant Inventory Quantity when importing products. github.com/medusajs/medusa/issues/11605
- GitHub Issue: Fail to import or export the products with available stocked inventory. github.com/medusajs/medusa/issues/9357
- GitHub Issue: Inventory items not created or updated on import. github.com/medusajs/medusa/issues/5981
On the solution:
- Medusa Documentation: Product Variant Inventory. docs.medusajs.com/resources/commerce-modules/product/variant-inventory
- Medusa Core Workflows Reference: importProductsWorkflow. docs.medusajs.com/resources/references/medusa-workflows/importProductsWorkflow
- Medusa Core Workflows Reference: updateInventoryLevelsWorkflow. docs.medusajs.com/resources/references/medusa-workflows/updateInventoryLevelsWorkflow
Stuck on a tricky one?
If you have a problem in Medusa storefront access, pricing, inventory, orders, or workflows 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 fix a catalog that looked right but would not sell?
If this saved you from re-keying stock counts by hand after a launch, 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