Shopify ERP and inventory sync in Python

Overview

Overselling, phantom stock and weekend reconciliation marathons are architecture problems, not data problems. The sync that avoids them rests on three decisions, made once:

• The ERP is the single source of truth for stock - Shopify displays quantities, it never owns them
• Publishing to Shopify is idempotent - you state absolute quantities, so re-running any sync after a failure is always safe
• A nightly reconciliation compares the two systems in full and repairs drift, because drift happens no matter how good the live path is

The failure modes this prevents are the classics of every ERP-Shopify thread: overselling, where Shopify shows stock that the ERP already committed to a B2B order; phantom stock, where a processed return takes days to appear on the storefront; and the manual reconciliation marathon, where someone compares spreadsheets every week because nobody trusts either system.

Zato sits between the two as the integration layer - the ERP side speaks whatever it speaks (OData for SAP and Dynamics 365, REST or SQL for others), the Shopify side is the GraphQL Admin API, and Python services in the middle own the mapping.

The live path

Two flows, one per direction, both thin.

ERP to Shopify: when stock changes in the ERP - a goods receipt, an adjustment, a B2B commitment - a service publishes the new absolute quantity with inventorySetQuantities. Absolute is the operative word: publishing "SKU HAT-001-S has 42 available at location X" is idempotent, while publishing deltas like "add 3" corrupts state on every retry or duplicate.

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class PublishInventory(Service):
    name = 'shopify.publish-inventory'

    input = 'quantities'

    def handle(self):

        conn = self.out.graphql['Shopify']

        # setQuantities states the absolute target -
        # publishing the same numbers twice is a no-op
        mutation = """
            mutation Publish($input: InventorySetQuantitiesInput!) {
                inventorySetQuantities(input: $input) {
                    userErrors { field message }
                }
            }
        """

        params = {'input': {
            'name': 'available',
            'reason': 'correction',
            'quantities': self.request.input.quantities,
        }}

        result = conn.execute(mutation, params=params)

        user_errors = result['inventorySetQuantities']['userErrors']

        if user_errors:
            raise Exception(f'Publish failed: {user_errors}')

Shopify to ERP: webhooks for orders/create feed sales into the ERP as they happen, deduplicated and processed asynchronously. Everything slow or batchy stays out of this path - the live flows move single facts quickly and nothing else.

Nightly reconciliation

A scheduler job runs while traffic is low: read all quantities from the ERP, read all quantities from Shopify with a bulk operation, compare per SKU and location, publish corrections for every mismatch and log each one. The corrections reuse the same idempotent publishing service as the live path - reconciliation is not a second write mechanism, it is the same one fed by a comparison.

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class ReconcileInventory(Service):
    """ Invoked by a scheduler job nightly, e.g. at 02:00.
    """
    name = 'shopify.reconcile-inventory'

    def handle(self):

        # Absolute quantities per SKU and location, from the ERP ..
        erp = self.invoke('erp.get-inventory-snapshot')

        # .. and from Shopify, exported earlier tonight in bulk -
        # each entry includes its quantity and inventory item GID
        shopify = self.invoke('shopify.get-inventory-snapshot')

        corrections = []

        for key, erp_quantity in erp.items():

            item = shopify[key]

            if item['quantity'] != erp_quantity:
                sku, location_id = key
                self.logger.warning('Drift for %s at %s: shopify=%s erp=%s',
                    sku, location_id, item['quantity'], erp_quantity)
                corrections.append({
                    'inventoryItemId': item['inventory_item_id'],
                    'locationId': location_id,
                    'quantity': erp_quantity,
                })

        if corrections:
            self.invoke('shopify.publish-inventory', {'quantities': corrections})

The log of mismatches is the health metric of the whole integration. A handful per night is normal drift from timing windows. A growing count names the SKUs and the direction - exactly where the live path is leaking - and if manual spreadsheet checks still consume hours each week, the reconciliation is not doing its job.

Pitfalls that break real syncs

When the design changes

The shape above survives growth - what changes is volume handling inside it. Catalog-wide publishes move from loops of mutations to bulk operations. Rate-limit pressure on the live path gets the backoff pattern from rate limits and backoff, applied in the one publishing service everything shares. And when oversell incidents recur or reconciliation keeps finding the same drift, the answer is architectural - a tighter live path or a second reconciliation window - never a human comparing counts by hand.

Learn more