product.py

1from automator.core.api import check_errors, shopify_request2from automator.core.errors import Retry, TaskAborted3from automator.core.models import Task, Variant4from automator.core.steps import Step56VENDOR = "carlomitchener"7CATEGORY = "gid://shopify/TaxonomyCategory/na"89MUTATION = """10mutation productSet($input: ProductSetInput!, $synchronous: Boolean!) {11    productSet(synchronous: $synchronous, input: $input) {12        product {13            id14            variants(first: 100) {15                nodes {16                    id17                    sku18                }19            }20        }21        userErrors {22            field23            message24        }25    }26}27"""2829def set_tags(task: Task) -> list[str]:30    return [task.product.category, task.product.title]3132def set_sizes(task: Task) -> list[str]:33    sizes = []34    for variant in task.variants:35        if variant.size not in sizes:36            sizes.append(variant.size)37    return sizes3839def set_product_options(task: Task) -> list[dict]:40    values = [{"name": size} for size in set_sizes(task)]41    return [{"name": "Size", "position": 1, "values": values}]4243def create_variant_payload(variant: Variant) -> dict:44    return {45        "inventoryItem": {"requiresShipping": True, "tracked": False},46        "optionValues": [{"optionName": "Size", "name": variant.size}],47        "price": variant.cost,48        "sku": variant.name,49        "taxable": False,50    }5152def create_product_payload(task: Task) -> dict:53    return {54        "category": CATEGORY,55        "descriptionHtml": "",56        "files": [{"id": m.shopify_id} for m in task.mockups],57        "handle": task.key,58        "productOptions": set_product_options(task),59        "productType": str(task.product.id),60        "status": "ACTIVE",61        "tags": set_tags(task),62        "title": task.product.title,63        "variants": [create_variant_payload(v) for v in task.variants],64        "vendor": VENDOR,65    }6667def create_product(task: Task) -> dict:68    variables = {"input": create_product_payload(task), "synchronous": True}69    result = shopify_request(task, MUTATION, variables)70    return check_errors(result, "productSet")7172def parse_result(task: Task, data: dict) -> Task:73    product = data["product"]74    task.product.shopify_id = product["id"]75    variants = {node["sku"]: node["id"] for node in product["variants"]["nodes"]}76    for variant in task.variants:77        if variant.name not in variants:78            raise TaskAborted(f"shopify lost variant {variant.name}")79        variant.shopify_id = variants[variant.name]80    return task8182def mrly_product(task: Task) -> Task:83    data = create_product(task)84    task = parse_result(task, data)85    task.place(Step.PING)86    raise Retry(f"Current: {Step.PRODUCT}. Next: {Step.PING}")