I am utilizing the Shopify GraphQL Admin API to retrieve product details based on SKUs. However, when querying a large number of SKUs, the request is limited to 250 results due to Shopify’s API constraints. Is there a recommended approach to efficiently handle bulk SKU queries beyond this limit?
Here is the code:-
def fetch_products_by_skus(skus):
if not skus or not isinstance(skus, str):
return None, [“Invalid SKUs input.”]
headers = {
"X-Shopify-Access-Token": ADMIN_API_TOKEN,
"Content-Type": "application/json",
"Accept": "application/json"
}
try:
query = """
query GetProductsBySkus($query: String!) {
productVariants(first: 250, query: $query) {
edges {
node {
sku
title
price
product {
title
productType
featuredImage {
url
}
variants(first: 100) {
edges {
node {
sku
title
price
metafield(namespace: "custom", key: "edition") {
value
}
}
}
}
metafields(first: 100) {
edges {
node {
namespace
key
value
}
}
}
}
}
}
}
}
"""
variables = {"query": skus}
response = requests.post(
f"https://{SHOPIFY_DOMAIN}/admin/api/{API_VERSION}/graphql.json",
json={'query': query, 'variables': variables},
headers=headers
)
response.raise_for_status()
data = response.json()
if safe_get(data, 'errors'):
errors = [err.get('message', 'Unknown error') for err in safe_get(data, 'errors', [])]
return None, errors
edges = safe_get(data, ['data', 'productVariants', 'edges'], [])
if not edges:
return None, ["No products found."]
products_by_sku = {}
errors = []
for edge in edges:
try:
node = safe_get(edge, 'node', {})
sku = safe_get(node, 'sku')
if not sku:
continue
product = safe_get(node, 'product', {})
metafield_edges = safe_get(product, ['metafields', 'edges'], [])
metafields = {}
for mf in metafield_edges:
mf_node = safe_get(mf, 'node', {})
namespace = safe_get(mf_node, 'namespace')
key = safe_get(mf_node, 'key')
value = safe_get(mf_node, 'value')
if namespace and key and value is not None:
metafields[f"{namespace}_{key}"] = value
variant_metafield = None
for v_edge in safe_get(product, ['variants', 'edges'], []):
v_node = safe_get(v_edge, 'node', {})
if safe_get(v_node, 'sku') == sku:
variant_metafield = safe_get(v_node, ['metafield', 'value'])
break
products_by_sku[sku] = {
'variant': node,
'product': product,
'metafields': metafields,
'edition': variant_metafield
}
except Exception as e:
errors.append(str(e))
return products_by_sku, errors
except requests.exceptions.RequestException as e:
return None, [str(e)]
except Exception as e:
return None, [str(e)]
