source-shopify-native: add ProductVariant Metafields

Add ProductVariantMetafields class with custom QUERY fragment including variants and metafields structure. Output is an individual variant records with metafields attached, using variant ID as primary key

Author: JustinASmith

source-shopify-native/acmeCo/flow.yaml

@@ -76,6 +76,10 @@ collections:
     schema: product_metafields.schema.yaml
     key:
       - /id
+  acmeCo/product_variant_metafields:
+    schema: product_variant_metafields.schema.yaml
+    key:
+      - /id
   acmeCo/product_variants:
     schema: product_variants.schema.yaml
     key:

source-shopify-native/acmeCo/product_variant_metafields.schema.yaml

@@ -0,0 +1,33 @@
+---
+$defs:
+  Meta:
+    properties:
+      op:
+        default: u
+        description: "Operation type (c: Create, u: Update, d: Delete)"
+        enum:
+          - c
+          - u
+          - d
+        title: Op
+        type: string
+      row_id:
+        default: -1
+        description: "Row ID of the Document, counting up from zero, or -1 if not known"
+        title: Row Id
+        type: integer
+    title: Meta
+    type: object
+additionalProperties: true
+properties:
+  _meta:
+    $ref: "#/$defs/Meta"
+    description: Document metadata
+  id:
+    title: Id
+    type: string
+required:
+  - id
+title: ShopifyGraphQLResource
+type: object
+x-infer-schema: true

source-shopify-native/source_shopify_native/graphql/__init__.py

@@ -16,6 +16,7 @@
 from .products.variants import ProductVariants
 from .products.media import ProductMedia
 from .products.metafields import ProductMetafields
+from .products.metafields import ProductVariantMetafields
 from .orders.agreements import OrderAgreements
 from .orders.fulfillment_orders import FulfillmentOrders
 from .orders.fulfillments import Fulfillments
@@ -43,6 +44,7 @@
     "ProductVariants",
     "ProductMedia",
     "ProductMetafields",
+    "ProductVariantMetafields",
     "OrderAgreements",
     "FulfillmentOrders",
     "Fulfillments",

source-shopify-native/source_shopify_native/graphql/products/metafields.py

@@ -1,6 +1,7 @@
 from datetime import datetime
 from logging import Logger
-from typing import AsyncGenerator
+import json
+from typing import Any, AsyncGenerator
 from ..metafields import MetafieldsResource
 
 
@@ -24,3 +25,121 @@ def process_result(
         return MetafieldsResource._process_metafields_result(
             log, lines, ProductMetafields.PARENT_ID_KEY
         )
+
+class ProductVariantMetafields(MetafieldsResource):
+    NAME = "product_variant_metafields"
+    PARENT_ID_KEY = "gid://shopify/ProductVariant/"
+    QUERY = """
+    variants {
+        edges {
+            node {
+                id
+                legacyResourceId
+                createdAt
+                updatedAt
+                metafields {
+                    edges {
+                        node {
+                            id
+                            legacyResourceId
+                            namespace
+                            key
+                            value
+                            type
+                            definition {
+                                id
+                                description
+                                namespace
+                                type {
+                                    category
+                                    name
+                                }
+                            }
+                            ownerType
+                            createdAt
+                            updatedAt
+                        }
+                    }
+                }
+            }
+        }
+    }
+    """
+
+    @staticmethod
+    def build_query(start: datetime, end: datetime) -> str:
+        return ProductVariantMetafields.build_query_with_fragment(
+            "products",
+            "UPDATED_AT",
+            start,
+            end,
+        )
+
+    @staticmethod
+    async def process_result(
+        log: Logger, lines: AsyncGenerator[bytes, None]
+    ) -> AsyncGenerator[dict, None]:
+        VARIANTS_KEY = "variants"
+        METAFIELDS_KEY = "metafields"
+        current_product = None
+        current_variant = None
+
+        async for line in lines:
+            record: dict[str, Any] = json.loads(line)
+            id: str = record.get("id", "")
+
+            if "gid://shopify/Product/" in id:
+                if current_product:
+                    # Yield any remaining variant from previous product
+                    if current_variant:
+                        yield current_variant
+                        current_variant = None
+                
+                current_product = record
+
+            elif "gid://shopify/ProductVariant/" in id:
+                if not current_product:
+                    log.error("Found a variant before finding a product.")
+                    raise RuntimeError()
+                elif record.get("__parentId", "") != current_product.get("id", ""):
+                    log.error(
+                        "Variant's parent id does not match the current product's id. Check if the JSONL response from Shopify is not ordered correctly.",
+                        {
+                            "variant.id": id,
+                            "variant.__parentId": record.get("__parentId"),
+                            "current_product.id": current_product.get("id"),
+                        },
+                    )
+                    raise RuntimeError()
+                
+                # Yield previous variant if exists
+                if current_variant:
+                    yield current_variant
+                
+                current_variant = record
+                current_variant[METAFIELDS_KEY] = []
+
+            elif "gid://shopify/Metafield/" in id:
+                if not current_variant:
+                    log.error("Found a metafield before finding a variant.")
+                    raise RuntimeError()
+                elif record.get("__parentId", "") != current_variant.get("id", ""):
+                    log.error(
+                        "Metafield's parent id does not match the current variant's id.",
+                        {
+                            "metafield.id": record.get("id"),
+                            "metafield.__parentId": record.get("__parentId"),
+                            "current_variant.id": current_variant.get("id"),
+                        },
+                    )
+                    raise RuntimeError()
+
+                current_variant[METAFIELDS_KEY].append(record)
+
+            else:
+                log.error("Unidentified line in JSONL response.", record)
+                raise RuntimeError()
+
+        # Yield final variant if exists
+        if current_variant:
+            yield current_variant

source-shopify-native/source_shopify_native/resources.py

@@ -37,6 +37,7 @@
     gql.ProductMedia,
     gql.ProductMetafields,
     gql.ProductVariants,
+    gql.ProductVariantMetafields,
     gql.FulfillmentOrders,
     gql.Fulfillments,
     gql.Orders,

source-shopify-native/test.flow.yaml

@@ -39,6 +39,10 @@ captures:
           name: product_variants
           interval: PT5M
         target: acmeCo/product_variants
+      - resource:
+          name: product_variant_metafields
+          interval: PT5M
+        target: acmeCo/product_variant_metafields
       - resource:
           name: fulfillment_orders
           interval: PT5M