Building a Shopify to Xero Invoice Sync in Python (With Code)

You've got a Shopify store. You've got Xero for accounting. And you're tired of paying £30/month for a connector app that occasionally breaks and gives you zero control over how invoices are mapped.

So let's build our own sync — in Python. I'll walk you through every piece: OAuth token management, fetching Shopify orders, mapping them to Xero invoices, handling VAT correctly, rate limiting, and state tracking so you never create duplicates.

This is production code. We've been running this at Beauty Hair Products for months without issues.

Related: If you want the higher-level business case first, check out Connect Shopify to Xero Without Paying for a Connector on our blog.

---

Architecture Overview

The flow is straightforward:

1. Fetch paid orders from Shopify's Admin API
2. Check state — skip orders we've already synced
3. Check Xero — skip invoices that already exist (belt and braces)
4. Map order → invoice — line items, VAT, shipping, discounts
5. POST to Xero — create the invoice
6. Update state — record what we've synced

No webhook complexity. No queue. Just a script you run on a cron schedule (we run ours every few hours).

---

Prerequisites

You'll need:

• Shopify Admin API token — create a private app with read_orders scope
• Xero OAuth2 app — registered at developer.xero.com
• Python 3.9+ with requests
• A Xero tenant (organisation) to push invoices into

pip install requests

---

1. Configuration

First, let's set up our constants. Store credentials in environment variables or a secrets manager — never hardcode them.

import os, json, time, logging, requests
from datetime import datetime, timezone
from pathlib import Path

# Shopify
SHOPIFY_STORE = os.environ["SHOPIFY_STORE"]  # e.g. "mystore.myshopify.com"
SHOPIFY_TOKEN = os.environ["SHOPIFY_TOKEN"]
SHOPIFY_API = f"https://{SHOPIFY_STORE}/admin/api/2024-01"
SHOPIFY_HEADERS = {"X-Shopify-Access-Token": SHOPIFY_TOKEN}

# Xero
XERO_CLIENT_ID = os.environ["XERO_CLIENT_ID"]
XERO_CLIENT_SECRET = os.environ["XERO_CLIENT_SECRET"]
XERO_TENANT_ID = os.environ["XERO_TENANT_ID"]
XERO_TOKEN_FILE = os.path.expanduser("~/.config/xero_tokens.json")
XERO_API = "https://api.xero.com/api.xro/2.0"

# Xero account codes — adjust to match YOUR chart of accounts
ACCOUNT_CODE_SALES = "4000"     # Sales revenue
ACCOUNT_CODE_SHIPPING = "4000"  # Or a separate shipping account

# State tracking
STATE_FILE = os.path.expanduser("~/.config/shopify_xero_state.json")

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("shopify-xero")

---

2. Xero OAuth2 Token Management

Xero uses OAuth2 with refresh tokens. Access tokens expire after 30 minutes, so we need automatic refresh:

def load_xero_tokens():
    """Load stored OAuth tokens from disk."""
    with open(XERO_TOKEN_FILE) as f:
        return json.load(f)

def save_xero_tokens(tokens):
    """Persist refreshed tokens."""
    with open(XERO_TOKEN_FILE, "w") as f:
        json.dump(tokens, f, indent=2)
    os.chmod(XERO_TOKEN_FILE, 0o600)  # Restrict permissions

def refresh_xero_token():
    """Refresh the Xero access token using the stored refresh token."""
    tokens = load_xero_tokens()
    r = requests.post("https://identity.xero.com/connect/token", data={
        "grant_type": "refresh_token",
        "refresh_token": tokens["refresh_token"],
        "client_id": XERO_CLIENT_ID,
        "client_secret": XERO_CLIENT_SECRET,
    })
    if r.status_code != 200:
        raise Exception(f"Token refresh failed: {r.status_code} {r.text[:200]}")

    new_tokens = r.json()
    save_xero_tokens(new_tokens)
    log.info("Xero token refreshed")
    return new_tokens["access_token"]

Key detail: chmod 0o600 on the token file. These tokens give full API access to your accounting data — treat them like passwords.

The Xero Request Wrapper

This handles token expiry AND rate limiting in one place:

def xero_headers(access_token):
    return {
        "Authorization": f"Bearer {access_token}",
        "Xero-Tenant-Id": XERO_TENANT_ID,
        "Content-Type": "application/json",
        "Accept": "application/json",
    }

def xero_request(method, endpoint, access_token, data=None, retry=True):
    """Make a Xero API request with auto-refresh and rate limit handling."""
    url = f"{XERO_API}/{endpoint}"
    headers = xero_headers(access_token)

    if method == "GET":
        r = requests.get(url, headers=headers)
    elif method == "POST":
        r = requests.post(url, headers=headers, json=data)
    else:
        raise ValueError(f"Unknown method: {method}")

    # Auto-refresh on 401
    if r.status_code == 401 and retry:
        log.info("Token expired, refreshing...")
        access_token = refresh_xero_token()
        return xero_request(method, endpoint, access_token, data, retry=False)

    # Back off on 429
    if r.status_code == 429:
        wait = int(r.headers.get("Retry-After", 5))
        log.warning(f"Rate limited, waiting {wait}s")
        time.sleep(wait)
        return xero_request(method, endpoint, access_token, data, retry=retry)

    return r, access_token

This is the single most valuable pattern in the whole script. Every Xero call goes through here, so you never have to think about token refresh or rate limits in your business logic.

---

3. Fetching Shopify Orders

Shopify's Admin API uses cursor-based pagination via Link headers:

def get_shopify_orders(since=None, limit=50):
    """Fetch paid orders from Shopify with pagination."""
    all_orders = []
    params = {
        "limit": min(limit, 250),
        "status": "any",
        "financial_status": "paid",
        "order": "created_at asc",
    }
    if since:
        params["created_at_min"] = since

    url = f"{SHOPIFY_API}/orders.json"
    while url and len(all_orders) < limit:
        r = requests.get(
            url, headers=SHOPIFY_HEADERS,
            params=params if "?" not in url else None
        )
        r.raise_for_status()
        orders = r.json().get("orders", [])
        all_orders.extend(orders)

        # Follow pagination
        link = r.headers.get("Link", "")
        url = None
        if 'rel="next"' in link:
            for part in link.split(","):
                if 'rel="next"' in part:
                    url = part.split("<")[1].split(">")[0]

        if len(orders) < 250:
            break
        time.sleep(0.5)  # Be nice to the API

    return all_orders[:limit]

Important: We filter by financial_status=paid because we only want to invoice orders that have actually been paid. No point creating an invoice for an abandoned checkout.

---

4. The Mapping: Shopify Order → Xero Invoice

This is where the business logic lives. Every business has different requirements here — ours maps to a consolidated contact (all Shopify revenue flows to one contact in Xero for cleaner reporting):

def order_to_xero_invoice(order):
    """Convert a Shopify order to a Xero invoice payload."""
    billing = order.get("billing_address", {}) or {}
    customer = order.get("customer", {}) or {}

    # Consolidated contact — all online sales under one roof
    contact = {
        "ContactID": os.environ.get("XERO_CONTACT_ID"),
        "Name": "Online Sales - Shopify",
    }

    # Customer name in the reference for traceability
    customer_name = (
        billing.get("name") or
        f"{customer.get('first_name', '')} {customer.get('last_name', '')}".strip() or
        customer.get("email", f"Customer {order['order_number']}")
    )

    # VAT handling: UK = 20% standard rate, international = zero-rated
    country = billing.get("country_code", "GB") if billing else "GB"
    tax_type = "OUTPUT2" if country == "GB" else "ZERORATEDOUTPUT"

    # Map line items
    line_items = []
    for li in order.get("line_items", []):
        unit_price = float(li["price"])
        quantity = li["quantity"]
        discount = float(li.get("total_discount", "0"))
        discount_pct = (
            (discount / (unit_price * quantity) * 100)
            if (unit_price * quantity) > 0 else 0
        )

        item = {
            "Description": li["title"],
            "Quantity": quantity,
            "UnitAmount": unit_price,
            "AccountCode": ACCOUNT_CODE_SALES,
            "TaxType": tax_type,
        }

        # Include SKU in description (not as ItemCode — Xero
        # rejects SKUs that aren't registered as inventory items)
        if li.get("sku"):
            item["Description"] += f" [{li['sku']}]"

        if discount_pct > 0:
            item["DiscountRate"] = round(discount_pct, 2)

        line_items.append(item)

    # Shipping as a line item
    for sl in order.get("shipping_lines", []):
        ship_price = float(sl.get("price", "0"))
        if ship_price > 0:
            line_items.append({
                "Description": f"Shipping: {sl.get('title', 'Delivery')}",
                "Quantity": 1,
                "UnitAmount": ship_price,
                "AccountCode": ACCOUNT_CODE_SHIPPING,
                "TaxType": tax_type,
            })

    return {
        "Type": "ACCREC",  # Sales invoice (Accounts Receivable)
        "Contact": contact,
        "Date": order["created_at"][:10],
        "DueDate": order["created_at"][:10],  # Already paid
        "LineAmountTypes": "Exclusive",  # Prices are ex-VAT
        "InvoiceNumber": f"SH-{order['name']}",
        "Reference": f"Shopify {order['name']} — {customer_name}",
        "Status": "AUTHORISED",
        "LineItems": line_items,
    }

VAT Handling Explained

This tripped me up initially. Key points:

• LineAmountTypes: "Exclusive" — Shopify prices are typically ex-VAT for B2C in the UK
• OUTPUT2 — Xero's code for 20% standard rate VAT (UK)
• ZERORATEDOUTPUT — for international orders (no UK VAT applies)
• We detect UK vs international from the billing address country code

If your Shopify prices are VAT-inclusive, change LineAmountTypes to "Inclusive".

The SKU Trap

Don't pass Shopify SKUs as Xero ItemCode. Xero will reject the invoice with a validation error if the SKU isn't registered as a tracked inventory item. Instead, append the SKU to the description for reference.

---

5. State Tracking (No Duplicates)

We track synced orders in a JSON state file — simple but effective:

def load_state():
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE) as f:
            return json.load(f)
    return {"last_synced_at": None, "synced_orders": []}

def save_state(state):
    with open(STATE_FILE, "w") as f:
        json.dump(state, f, indent=2)

The sync function checks two layers:

1. Local state — fast, no API call
2. Xero invoice lookup — belt and braces, catches invoices created manually

# In the sync loop:
synced = set(state.get("synced_orders", []))

for order in orders:
    order_name = order["name"]
    invoice_num = f"SH-{order_name}"

    # Layer 1: Local state
    if order_name in synced:
        skipped += 1
        continue

    # Layer 2: Check Xero
    r, access_token = xero_request(
        "GET", f"Invoices?InvoiceNumbers={invoice_num}", access_token
    )
    if r.status_code == 200:
        if r.json().get("Invoices", []):
            synced.add(order_name)
            skipped += 1
            continue

    # ... create the invoice ...

We also cap the state file at 5,000 order names to prevent it growing forever:

state["synced_orders"] = list(synced)[-5000:]

---

6. Rate Limiting

Both APIs have rate limits:

• Shopify: 2 requests/second (leaked bucket)
• Xero: 60 requests/minute per tenant

Our approach is simple: time.sleep(0.5) between requests. Not sophisticated, but reliable. The xero_request wrapper handles 429 responses with proper Retry-After backoff.

---

7. Putting It All Together

The main sync function:

def sync_orders(since=None, limit=50, dry_run=False):
    state = load_state()
    since = since or state.get("last_synced_at")

    if not since:
        log.error("No start date. Use --since YYYY-MM-DD")
        return

    orders = get_shopify_orders(since=since, limit=limit)
    log.info(f"Found {len(orders)} paid orders")

    tokens = load_xero_tokens()
    access_token = tokens["access_token"]
    synced = set(state.get("synced_orders", []))

    created = skipped = errors = 0

    for order in orders:
        order_name = order["name"]
        invoice_num = f"SH-{order_name}"

        if order_name in synced:
            skipped += 1
            continue

        # Check Xero for existing invoice
        r, access_token = xero_request(
            "GET", f"Invoices?InvoiceNumbers={invoice_num}", access_token
        )
        if r.status_code == 200 and r.json().get("Invoices"):
            synced.add(order_name)
            skipped += 1
            continue

        # Create invoice
        invoice_data = order_to_xero_invoice(order)

        if dry_run:
            log.info(f"[DRY RUN] {invoice_num}: £{order['total_price']}")
            created += 1
            continue

        r, access_token = xero_request(
            "POST", "Invoices", access_token,
            {"Invoices": [invoice_data]}
        )

        if r.status_code in (200, 201):
            inv = r.json().get("Invoices", [{}])[0]
            if inv.get("HasErrors"):
                log.error(f"Validation error: {inv.get('ValidationErrors')}")
                errors += 1
            else:
                log.info(f"Created {invoice_num} -> {inv.get('InvoiceID')}")
                synced.add(order_name)
                created += 1
        else:
            log.error(f"Failed {invoice_num}: {r.status_code}")
            errors += 1

        time.sleep(0.5)

    # Update state
    if not dry_run:
        state["last_synced_at"] = orders[-1]["created_at"] if orders else since
        state["synced_orders"] = list(synced)[-5000:]
        save_state(state)

    log.info(f"Done: {created} created, {skipped} skipped, {errors} errors")

---

Running It

# First run — specify a start date
python3 shopify_xero_sync.py --sync --since 2026-01-01

# Subsequent runs use saved state
python3 shopify_xero_sync.py --sync

# Dry run to preview
python3 shopify_xero_sync.py --sync --dry-run

# Backfill a date range
python3 shopify_xero_sync.py --sync --since 2025-06-01 --until 2025-12-31

# Test connections
python3 shopify_xero_sync.py --test

Cron Schedule

We run ours every 4 hours:

0 */4 * * * cd ~/scripts && python3 shopify_xero_sync.py --sync --limit 100 >> logs/sync.log 2>&1

---

Lessons from Production

After running this for several months, here's what we learned:

1. Don't trust ItemCode — Xero validates SKUs against its inventory. Just put them in the description.

2. Consolidated contacts save headaches — Creating a new Xero contact per Shopify customer creates thousands of contacts. Use one "Online Sales" contact and put the customer name in the Reference field.

3. Two layers of duplicate detection — Local state is fast but can drift. Checking Xero directly is slower but authoritative. Do both.

4. LineAmountTypes matters enormously — Getting this wrong means your VAT calculations are off. Test with a known order and verify the totals match.

5. Keep state files bounded — We learned this one the hard way when our state file hit 50MB. The [-5000:] slice keeps it manageable.

6. Rate limit proactively — Don't wait for 429s. A sleep(0.5) between requests costs you almost nothing but prevents painful backoff cascades.

---

What This Doesn't Do (Yet)

• Refund handling — We handle these manually in Xero for now
• Webhook-driven sync — Possible with Shopify webhooks, but polling is simpler to maintain
• Multi-currency — All our sales are in GBP. You'd need to add currency mapping for international stores

---

Wrapping Up

Total cost of this solution: £0/month. Total lines of code: ~300. Total control: 100%.

The connector apps charge £20-40/month and give you a config screen. This gives you code you can read, modify, and debug. When Xero changes their VAT types or Shopify changes their order format, you fix it in 5 minutes instead of waiting for a third-party update.

If you're running a UK e-commerce business on Shopify + Xero, I'd genuinely recommend this approach over any paid connector. The initial setup takes an afternoon, and then it just works.

---

Built and maintained by Drakon Systems — we build automation tools for small businesses tired of paying for things that should be free.