# Shopify bulk operations from Python Starting bulk queries, polling their status with the scheduler and processing the JSONL results. ## Overview {#overview} Above a few thousand records, paginated loops are throttled by rate limits. A bulk operation submits 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 {#starting-a-bulk-operation} The inner query has no pagination - Shopify iterates over the whole catalog server-side. ```python # -*- 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 {#polling-with-the-scheduler} Poll with the [scheduler](https://zato.io/docs/dev/examples/scheduler.html) - a job that invokes a status-check service every few minutes. No thread is held while the operation runs, and a job that takes eight hours costs the same code as one that takes five minutes. The service reads currentBulkOperation and acts on the status it reports - still running means do nothing until the next tick. ```python # -*- 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 {#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. ```python # -*- 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 {#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 use rate-limited per-object calls for unsupported mutations, as described in [rate limits and backoff](https://zato.io/docs/dev/shopify/rate-limits-and-backoff.html) where needed. ## What goes wrong in production {#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. ## Learn more {#learn-more} - [Shopify integration overview](https://zato.io/integrations/shopify/) - Start here to see how teams build Shopify integrations - [The GraphQL Admin API](https://zato.io/docs/dev/shopify/graphql-admin-api-python.html) - Connections, access tokens, API versions, queries and mutations - [Products and variants](https://zato.io/docs/dev/shopify/products-and-variants.html) - productSet, productVariantsBulkUpdate and the 2048-variant model - [Rate limits and backoff](https://zato.io/docs/dev/shopify/rate-limits-and-backoff.html) - The cost-based leaky bucket, the 1,000-point ceiling and THROTTLED - [Reliable webhooks](https://zato.io/docs/dev/shopify/webhooks-python.html) - HMAC verification, fast acknowledgments and deduplication - [GraphQL error handling](https://zato.io/docs/dev/shopify/graphql-error-handling.html) - Why 200 OK does not mean success - errors and userErrors - [ERP and inventory sync](https://zato.io/docs/dev/shopify/erp-inventory-sync.html) - Source of truth, idempotent publishing and nightly reconciliation