Shopify GraphQL批量图片上传遇字符串格式化KeyError求助

问题描述 #

我正在尝试通过Shopify GraphQL API实现约2000张图片的自动化批量上传，需基于SKU/ProductID匹配确保图片对应正确商品。此前REST API版本可正常运行，但改用GraphQL提升效率后，多次调整请求格式仍触发KeyError错误。我是GraphQL和Shopify新手，附上相关代码及错误栈，请求定位问题。

代码示例 #

import shopify
import base64
import os

# 初始化Shopify客户端
shopify.ShopifyResource.set_site("https://your-shop-name.myshopify.com/admin/api/2024-07")
shopify.ShopifyResource.set_oauth_credentials(api_key="your-api-key", secret="your-secret")

def upload_product_image(product_id, image_path):
    query = '''
    mutation productImageCreate($input: ProductImageInput!) {
        productImageCreate(input: $input) {
            product {
                id
                title
            }
            userErrors {
                field
                message
            }
        }
    }
    '''
    with open(image_path, 'rb') as img_file:
        image_data = img_file.read()
    variables = {
        "input": {
            "productId": product_id,
            "image": {
                "src": f"data:image/jpeg;base64,{base64.b64encode(image_data).decode('utf-8')}"
            }
        }
    }
    response = shopify.GraphQL().execute(query, variables=variables)
    # 错误触发位置
    return response['data']['productImageCreate']['product']

# 批量处理逻辑（假设image_sku_map为SKU与图片路径的映射字典）
for sku, image_path in image_sku_map.items():
    product = shopify.Product.find(sku=sku)
    if product:
        upload_product_image(product.id, image_path)

错误栈 #

KeyError                                  Traceback (most recent call last)
<ipython-input-12-abc123> in <module>
     28     product = shopify.Product.find(sku=sku)
     29     if product:
---> 30         upload_product_image(product.id, image_path)

<ipython-input-12-abc123> in upload_product_image(product_id, image_path)
     22     response = shopify.GraphQL().execute(query, variables=variables)
     23     # 错误触发位置
---> 24     return response['data']['productImageCreate']['product']

KeyError: 'productImageCreate'

排查与解决步骤 #

1. 先处理响应中的错误信息 #

Shopify GraphQL API如果请求存在问题，data字段下不会返回productImageCreate，而是会在userErrors中返回错误详情。需要先判断响应结构：

def upload_product_image(product_id, image_path):
    # ... 原有查询与变量代码 ...
    response = shopify.GraphQL().execute(query, variables=variables)
    response_data = response.get('data', {})
    
    # 先检查是否存在错误
    mutation_result = response_data.get('productImageCreate', {})
    if 'userErrors' in mutation_result and mutation_result['userErrors']:
        for error in mutation_result['userErrors']:
            print(f"上传图片到商品 {product_id} 失败: {error['message']} (字段: {error['field']})")
        return None
    
    # 安全获取返回的商品信息
    product = mutation_result.get('product')
    if not product:
        print(f"商品 {product_id} 未返回有效数据")
        return None
    return product

2. 修正ProductID格式 #

GraphQL要求使用全局ID格式（gid://shopify/Product/[数字ID]），而非REST API返回的纯数字ID。需转换ID格式：

# 在批量处理逻辑中替换ID格式
for sku, image_path in image_sku_map.items():
    product = shopify.Product.find(sku=sku)
    if product:
        # 转换为GraphQL兼容的全局ID
        graphql_product_id = f"gid://shopify/Product/{product.id}"
        upload_product_image(graphql_product_id, image_path)

3. 验证图片Base64编码 #

确保图片编码完整且格式正确，可单独测试单张图片的编码是否能被Shopify识别：

# 测试编码是否正常
with open(image_path, 'rb') as img_file:
    image_data = img_file.read()
base64_str = base64.b64encode(image_data).decode('utf-8')
# 检查编码长度是否合理，避免空值或截断
print(f"图片编码长度: {len(base64_str)}")

4. 处理速率限制 #

Shopify GraphQL API默认限制每分钟1000次查询，批量上传2000张图片需添加重试机制避免429错误：

import time

def upload_product_image(product_id, image_path):
    # ... 原有查询与变量代码 ...
    try:
        response = shopify.GraphQL().execute(query, variables=variables)
    except shopify.exceptions.HttpError as e:
        if e.response.status_code == 429:
            retry_after = int(e.response.headers.get('Retry-After', 10))
            print(f"触发速率限制，{retry_after}秒后重试...")
            time.sleep(retry_after)
            return upload_product_image(product_id, image_path)
        else:
            raise
    # ... 后续响应处理代码 ...

内容的提问来源于stack exchange，提问作者Bobby Pope