"""Pull Shopify product attributes/metafields/options into a local registry. CLI: python -m app.jobs.shopify_attribute_sync Env: SHOPIFY_SHOP_DOMAIN=example.myshopify.com SHOPIFY_ADMIN_ACCESS_TOKEN=shpat_... # or, for Dev Dashboard apps installed on owned stores: SHOPIFY_CLIENT_ID=... SHOPIFY_CLIENT_SECRET=... SHOPIFY_API_VERSION=2026-04 SHOPIFY_PRODUCT_OPTIONS_SAMPLE_SIZE=250 """ from __future__ import annotations import argparse import logging import sys from datetime import datetime, timezone from typing import Any, Dict, List, Optional from db.source_imports import normalize_column_key logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) STANDARD_PRODUCT_FIELDS = [ ("product", "sku", "SKU", "string"), ("product", "title", "Title", "string"), ("product", "description", "Description", "html"), ("product", "vendor", "Vendor", "string"), ("product", "product_type", "Product type", "string"), ("product", "handle", "Handle", "string"), ("product", "status", "Status", "string"), ("product", "tags", "Tags", "list.single_line_text_field"), ("product", "seo_title", "SEO title", "string"), ("product", "seo_description", "SEO description", "string"), ("variant", "price", "Price", "money"), ("variant", "compare_at_price", "Compare-at price", "money"), ("variant", "barcode", "Barcode", "string"), ("variant", "inventory_quantity", "Inventory quantity", "number_integer"), ("variant", "inventory_policy", "Inventory policy", "string"), ("variant", "taxable", "Taxable", "boolean"), ("variant", "weight", "Weight", "number_decimal"), ("variant", "weight_unit", "Weight unit", "string"), ] METAFIELD_DEFINITIONS_QUERY = """ query MetafieldDefinitions($ownerType: MetafieldOwnerType!, $after: String) { metafieldDefinitions(first: 250, ownerType: $ownerType, after: $after) { pageInfo { hasNextPage endCursor } nodes { id name namespace key ownerType type { name category } description validations { name value } } } } """ PRODUCT_OPTIONS_QUERY = """ query ProductOptions($first: Int!, $after: String) { products(first: $first, after: $after) { pageInfo { hasNextPage endCursor } nodes { id title options { id name position values } } } } """ def run_shopify_attribute_sync( *, connection_id: Optional[int] = None, shop_code: Optional[str] = None, session=None, ) -> Dict[str, Any]: from db.compat_connections import SHOPIFY_COMPAT_ID_OFFSET, decode_compat_connection_id from db.models import ShopifyConnection from db.session import get_session from db.shopify_registry import alternate_registry_shop_codes, resolve_registry_shop_code, resolve_shopify_connection from db.shopify_repositories import SqlAlchemyShopifyAttributeRegistryRepository from settings import load_shopify_config from shopify.connections import build_client own_session = session is None ctx = get_session() if own_session else _noop_ctx(session) with ctx as sess: connection: Optional[ShopifyConnection] = None if connection_id: connection = resolve_shopify_connection(sess, connection_id=connection_id) if connection is None: channel_type = "unknown" native_id = connection_id if connection_id >= SHOPIFY_COMPAT_ID_OFFSET: channel_type, native_id = decode_compat_connection_id(connection_id) if channel_type != "shopify": return { "status": "failed", "error": f"Connection {connection_id} is not a Shopify connection", } return { "status": "failed", "error": f"Shopify connection {connection_id} not found (native id {native_id})", } else: connection = resolve_shopify_connection(sess, shop_code=shop_code) cfg = load_shopify_config() client = build_client(connection) effective_shop_code = resolve_registry_shop_code(sess, connection=connection, shop_code=shop_code) if not effective_shop_code: return {"status": "failed", "error": "Shopify shop_code is required (connection or SHOPIFY_SHOP_CODE)"} api_version = (connection.api_version if connection else None) or cfg.api_version options_sample_size = cfg.product_options_sample_size if not client.shop_domain: return {"status": "failed", "error": "Shopify shop_domain is required (connection or SHOPIFY_SHOP_DOMAIN)"} fetched_at = datetime.now(timezone.utc) rows: List[Dict[str, Any]] = [] rows.extend(_standard_field_rows()) product_metafields = _fetch_metafield_definitions(client, "PRODUCT") variant_metafields = _fetch_metafield_definitions(client, "PRODUCTVARIANT") rows.extend(_metafield_rows(product_metafields)) rows.extend(_metafield_rows(variant_metafields)) option_rows = _fetch_product_option_rows(client, options_sample_size) rows.extend(option_rows) repo = SqlAlchemyShopifyAttributeRegistryRepository(sess) consolidation = repo.consolidate_shop_codes( effective_shop_code, alternate_registry_shop_codes(connection=connection, canonical_shop_code=effective_shop_code), ) count = repo.bulk_replace(effective_shop_code, rows, fetched_at) if own_session: sess.commit() counts = { "standard_fields": len(STANDARD_PRODUCT_FIELDS), "product_metafields": len(product_metafields), "variant_metafields": len(variant_metafields), "product_options": len(option_rows), "registry_rows": count, } return { "status": "success", "shop_code": effective_shop_code, "connection_id": connection.id if connection is not None else None, "api_version": api_version, "consolidation": consolidation, "counts": counts, } def _standard_field_rows() -> List[Dict[str, Any]]: rows = [] for owner_type, code, name, data_type in STANDARD_PRODUCT_FIELDS: rows.append( { "owner_type": owner_type, "attribute_code": normalize_column_key(code), "name": name, "data_type": data_type, "is_standard": True, "raw_json": {"source": "standard_field"}, } ) return rows def _fetch_metafield_definitions(client: Any, owner_type: str) -> List[Dict[str, Any]]: rows: List[Dict[str, Any]] = [] after: Optional[str] = None while True: data = client.graphql(METAFIELD_DEFINITIONS_QUERY, {"ownerType": owner_type, "after": after}) connection = data.get("metafieldDefinitions") or {} nodes = connection.get("nodes") or [] rows.extend([node for node in nodes if isinstance(node, dict)]) page_info = connection.get("pageInfo") or {} if not page_info.get("hasNextPage"): break after = page_info.get("endCursor") if not after: break return rows def _metafield_rows(definitions: List[Dict[str, Any]]) -> List[Dict[str, Any]]: rows = [] for definition in definitions: namespace = str(definition.get("namespace") or "").strip() key = str(definition.get("key") or "").strip() owner_type = str(definition.get("ownerType") or "").strip().lower() if not namespace or not key or not owner_type: continue type_obj = definition.get("type") or {} rows.append( { "owner_type": f"{owner_type}_metafield", "attribute_code": normalize_column_key(f"{namespace}_{key}"), "name": definition.get("name") or key, "namespace": namespace, "key": key, "data_type": type_obj.get("name") if isinstance(type_obj, dict) else None, "is_standard": False, "raw_json": definition, } ) return rows def _fetch_product_option_rows(client: Any, sample_size: int) -> List[Dict[str, Any]]: if sample_size <= 0: return [] option_map: Dict[str, Dict[str, Any]] = {} remaining = sample_size after: Optional[str] = None while remaining > 0: first = min(250, remaining) data = client.graphql(PRODUCT_OPTIONS_QUERY, {"first": first, "after": after}) connection = data.get("products") or {} for product in connection.get("nodes") or []: if not isinstance(product, dict): continue for option in product.get("options") or []: if not isinstance(option, dict): continue name = str(option.get("name") or "").strip() code = normalize_column_key(f"option_{name}") if not code: continue row = option_map.setdefault( code, { "owner_type": "product_option", "attribute_code": code, "name": name, "data_type": "option", "is_standard": True, "raw_json": {"sample_product_ids": [], "values": []}, }, ) raw = row["raw_json"] product_id = product.get("id") if product_id and len(raw["sample_product_ids"]) < 10 and product_id not in raw["sample_product_ids"]: raw["sample_product_ids"].append(product_id) for value in option.get("values") or []: if value not in raw["values"]: raw["values"].append(value) page_info = connection.get("pageInfo") or {} if not page_info.get("hasNextPage"): break after = page_info.get("endCursor") if not after: break remaining -= first return list(option_map.values()) class _noop_ctx: def __init__(self, obj): self._obj = obj def __enter__(self): return self._obj def __exit__(self, *args): pass def main() -> int: parser = argparse.ArgumentParser(description="Pull Shopify attributes/metafields/options into registry") parser.add_argument("--connection-id", type=int, default=None, help="Shopify connection id (native or compat)") parser.add_argument("--shop-code", default=None, help="Shopify shop_code override") args = parser.parse_args() result = run_shopify_attribute_sync(connection_id=args.connection_id, shop_code=args.shop_code) if result.get("status") != "success": print(result.get("error", "Shopify attribute sync failed"), file=sys.stderr) return 1 counts = result.get("counts", {}) print( "Shopify attribute sync complete: " f"shop_code={result.get('shop_code')}, " f"standard_fields={counts.get('standard_fields', 0)}, " f"product_metafields={counts.get('product_metafields', 0)}, " f"variant_metafields={counts.get('variant_metafields', 0)}, " f"product_options={counts.get('product_options', 0)}, " f"registry_rows={counts.get('registry_rows', 0)}" ) return 0 if __name__ == "__main__": sys.exit(main())