Shopify bulk operations from Python
Starting bulk queries, polling their status with the scheduler and processing the JSONL results.
Overview
Above a few thousand records, paginated loops lose to rate limits. Bulk operations hand the whole job to Shopify instead, in three steps that never block your code:
- You submit a query wrapped in bulkOperationRunQuery and Shopify starts a background job on its side, immune to rate limits
- You ask for the job's status with the currentBulkOperation query - not continuously, but on a schedule
- When the status reaches COMPLETED, the response contains a signed URL to a JSONL file with the full result, one JSON object per line, which your service downloads and processes
The economics are why this matters: a full catalog read through paginated queries burns rate-limit points for hours, while the same read as a bulk operation costs the one mutation that started it. The trade is latency - a bulk job takes minutes to hours - so bulk operations belong to syncs, exports and backfills, not to request-response paths.
Starting a bulk operation
The inner query has no pagination - Shopify walks the whole catalog itself.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class StartProductExport(Service):
name = 'shopify.start-product-export'
def handle(self):
conn = self.out.graphql['Shopify']
mutation = """
mutation {
bulkOperationRunQuery(
query: \"\"\"
{
products {
edges {
node {
id
title
variants {
edges { node { id sku } }
}
}
}
}
}
\"\"\"
) {
bulkOperation { id status }
userErrors { field message }
}
}
"""
result = conn.execute(mutation)
operation = result['bulkOperationRunQuery']
if operation['userErrors']:
raise Exception(f'Could not start: {operation["userErrors"]}')
self.logger.info('Started %s', operation['bulkOperation']['id'])
Polling with the scheduler
Poll with the scheduler - a job that invokes a status-check service every few minutes. No thread sleeps anywhere, and a job that takes eight hours costs the same code as one that takes five minutes. The service reads currentBulkOperation and acts on what it sees - still running means do nothing until the next tick.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class CheckBulkStatus(Service):
""" Invoked by a scheduler job every few minutes.
"""
name = 'shopify.check-bulk-status'
def handle(self):
conn = self.out.graphql['Shopify']
query = """
{
currentBulkOperation {
id
status
errorCode
objectCount
url
}
}
"""
result = conn.execute(query)
operation = result['currentBulkOperation']
# No operation has run yet on this shop
if operation is None:
return
status = operation['status']
if status == 'RUNNING':
self.logger.info('Still running, %s objects so far', operation['objectCount'])
elif status == 'COMPLETED':
self.invoke_async('shopify.process-bulk-result', {'url': operation['url']})
else:
# FAILED, CANCELED or EXPIRED - alert and restart deliberately
self.logger.warning('Bulk operation %s: %s (%s)',
operation['id'], status, operation['errorCode'])
Processing the JSONL result
Process the file line by line - each line is one complete JSON object, so even a multi-gigabyte result never needs to fit in memory as a whole. Nested connections arrive flattened: child objects such as variants appear as their own lines, each with a __parentId field pointing at the product line they belong to.
# -*- coding: utf-8 -*-
# stdlib
from json import loads
# requests
import requests
# Zato
from zato.server.service import Service
class ProcessBulkResult(Service):
name = 'shopify.process-bulk-result'
input = 'url'
def handle(self):
# The URL is signed by Shopify and valid for a week
response = requests.get(self.request.input.url, stream=True)
products = {}
for line in response.iter_lines():
item = loads(line)
# Child objects point at their parent line
if '__parentId' in item:
parent = products[item['__parentId']]
parent['variants'].append(item)
else:
item['variants'] = []
products[item['id']] = item
self.logger.info('Loaded %s products', len(products))
Bulk writes
bulkOperationRunMutation runs one mutation, such as productSet, across many inputs. You upload a JSONL file of input objects through Shopify's stagedUploadsCreate flow, then start the operation with the mutation string and the upload's key. The same status lifecycle and the same scheduler polling apply.
Not every mutation is supported in bulk - check the current list before designing around it, and fall back to rate-limited per-object calls as described in rate limits and backoff where needed.
What goes wrong in production
Three things, all reported repeatedly by teams running nightly syncs:
- Duration varies on Shopify's side - the same export that takes an hour one week can take eight the next, with no change on yours, so schedule syncs with slack and alert on wall-clock age rather than assuming a fixed window. The hard ceiling is 10 days, after which Shopify fails the job - healthy jobs finish in minutes to hours, so alert well before that.
- Jobs fail or get canceled without much explanation - the status service above already routes FAILED, CANCELED and EXPIRED to a warning, and the right response is a deliberate restart, not an automatic tight retry.
- Only one bulk operation runs per shop at a time - a stuck or forgotten job blocks every later one, so if currentBulkOperation shows something unexpected in CREATED or RUNNING, bulkOperationCancel clears the slot. The limit is per shop per app, so each shop's connection can have its own operation running concurrently.
A running bulk operation does not consume your rate-limit points - only the start mutation and the status polls cost points, both negligible.