source-shopify-native: migrate legacy state before Pydantic validation

Add ShopifyOpen custom request class with model_validator that transforms
legacy flat state to dict-based format BEFORE Pydantic parsing. This prevents
hybrid state from being created when Flow's checkpoint merging combines old
flat state with new dict-based state.

Previously, state migration happened at runtime via checkpoint, and Flow's
merge behavior would create hybrid state like:
  {"inc": {"cursor": "...", "store-id": {"cursor": "..."}}}
which Pydantic cannot parse.

Now the migration happens in the model_validator, so the state is already
dict-based when Pydantic parses it, and no format-changing checkpoint is
ever created.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: JustinASmith

source-shopify-native/source_shopify_native/__init__.py

@@ -1,13 +1,12 @@
 from logging import Logger
-from typing import Callable, Awaitable
+from typing import Callable, Awaitable, Union
 
 from estuary_cdk.flow import (
     ConnectorSpec,
     ValidationError,
 )
 from estuary_cdk.capture import (
     BaseCaptureConnector,
-    Request,
     Task,
     common,
     request,
@@ -20,9 +19,21 @@
     ConnectorState,
     EndpointConfig,
     OAUTH2_SPEC,
+    ShopifyOpen,
 )
 
 
+# Custom Request type that uses ShopifyOpen for cleaning hybrid state during parsing.
+# This is necessary because Flow's checkpoint merging can create hybrid state when
+# migrating from flat to dict-based format, which Pydantic cannot parse.
+ShopifyRequest = Union[
+    request.Spec,
+    request.Discover[EndpointConfig],
+    request.Validate[EndpointConfig, ResourceConfig],
+    ShopifyOpen,
+]
+
+
 def _bindings_use_multi_store_keys(bindings: list | None) -> bool:
     """Check if any binding uses the multi-store key structure (/_meta/store in key)."""
     if not bindings:
@@ -159,7 +170,7 @@ class Connector(
     BaseCaptureConnector[EndpointConfig, ResourceConfig, ConnectorState],
 ):
     def request_class(self):
-        return Request[EndpointConfig, ResourceConfig, ConnectorState]
+        return ShopifyRequest
 
     async def spec(self, log: Logger, _: request.Spec) -> ConnectorSpec:
         return ConnectorSpec(

source-shopify-native/source_shopify_native/models.py

@@ -19,11 +19,13 @@
     BaseDocument,
     LongLivedClientCredentialsOAuth2Credentials,
     OAuth2Spec,
+    ResourceConfig,
     ResourceState,
 )
 from estuary_cdk.capture.common import (
     ConnectorState as GenericConnectorState,
 )
+from estuary_cdk.capture import request
 from .graphql.common import dt_to_str, str_to_dt
 
 
@@ -552,3 +554,120 @@ def get_page_info(self) -> PageInfo:
     )
 
     return DataModel
+
+
+def _is_legacy_flat_state(resource_state: dict[str, Any]) -> bool:
+    """Check if a resource state is in legacy flat format.
+
+    Legacy flat format has cursor/next_page directly in inc/backfill,
+    not nested under a store_id key.
+    """
+    inc = resource_state.get("inc")
+    if isinstance(inc, dict):
+        # Legacy flat: {"cursor": "..."} with no nested dicts containing cursor
+        if "cursor" in inc:
+            has_store_keys = any(
+                isinstance(v, dict) and "cursor" in v
+                for k, v in inc.items() if k != "cursor"
+            )
+            if not has_store_keys:
+                return True
+    return False
+
+
+def _migrate_resource_state_to_dict(resource_state: dict[str, Any], store_id: str) -> None:
+    """Migrate a resource state from legacy flat format to dict-based format.
+
+    Legacy format: {"inc": {"cursor": "..."}, "backfill": {"next_page": "...", "cutoff": "..."}}
+    Dict format: {"inc": {"store_id": {"cursor": "..."}}, "backfill": {"store_id": {...}}}
+
+    This transforms the state IN PLACE before Pydantic validation, preventing
+    hybrid state from ever being created via checkpoint merging.
+
+    Args:
+        resource_state: The raw resource state dict to migrate (modified in place).
+        store_id: The store ID to use as the dict key.
+    """
+    inc = resource_state.get("inc")
+    if isinstance(inc, dict) and "cursor" in inc:
+        # Check this is pure legacy (no store keys already present)
+        has_store_keys = any(
+            isinstance(v, dict) and "cursor" in v
+            for k, v in inc.items() if k != "cursor"
+        )
+        if not has_store_keys:
+            # Pure legacy - wrap in store_id dict
+            resource_state["inc"] = {store_id: inc}
+
+    backfill = resource_state.get("backfill")
+    if isinstance(backfill, dict) and "next_page" in backfill:
+        # Check this is pure legacy (no store keys already present)
+        has_store_keys = any(
+            isinstance(v, dict) and "next_page" in v
+            for k, v in backfill.items() if k not in ("next_page", "cutoff")
+        )
+        if not has_store_keys:
+            # Pure legacy - wrap in store_id dict
+            resource_state["backfill"] = {store_id: backfill}
+
+
+def _get_store_id_from_config(config_data: dict[str, Any]) -> str | None:
+    """Extract the store ID from config data for state migration.
+
+    Legacy captures have exactly one store, so we use that store's ID
+    as the key when migrating from flat to dict-based state.
+    """
+    # New multi-store format: {"stores": [{"store": "store-id", ...}]}
+    stores = config_data.get("stores", [])
+    if stores and isinstance(stores, list) and len(stores) > 0:
+        first_store = stores[0]
+        if isinstance(first_store, dict):
+            return first_store.get("store")
+
+    # Legacy single-store format: {"store": "store-id", ...}
+    return config_data.get("store")
+
+
+class ShopifyOpen(request.Open["EndpointConfig", ResourceConfig, ConnectorState]):
+    """Custom Open request that migrates legacy state before Pydantic validation.
+
+    This transforms legacy flat state to dict-based format BEFORE Pydantic parsing,
+    which prevents hybrid state from ever being created. Without this, the migration
+    would happen at runtime via checkpoint, and Flow's checkpoint merging would
+    create hybrid state that Pydantic cannot parse.
+    """
+
+    @model_validator(mode="before")
+    @classmethod
+    def migrate_legacy_state(cls, data: Any) -> Any:
+        """Migrate legacy flat state to dict-based format before Pydantic validation."""
+        if not isinstance(data, dict):
+            return data
+
+        # Get store_id from config
+        capture = data.get("capture", {})
+        if not isinstance(capture, dict):
+            return data
+
+        config_data = capture.get("config", {})
+        if not isinstance(config_data, dict):
+            return data
+
+        store_id = _get_store_id_from_config(config_data)
+        if not store_id:
+            return data
+
+        # Migrate each binding's state
+        state = data.get("state", {})
+        if not isinstance(state, dict):
+            return data
+
+        binding_states = state.get("bindingStateV1", {})
+        if not isinstance(binding_states, dict):
+            return data
+
+        for resource_state in binding_states.values():
+            if isinstance(resource_state, dict):
+                _migrate_resource_state_to_dict(resource_state, store_id)
+
+        return data

source-shopify-native/source_shopify_native/resources.py

@@ -161,62 +161,44 @@ def _reconcile_connector_state(
 ) -> None:
     """Reconcile connector state to ensure all stores have proper state entries.
 
-    This handles:
-    1. Legacy flat state migration to dict-based state
-    2. Adding new stores to existing dict-based state
+    This handles adding new stores to existing dict-based state. Legacy flat state
+    migration is handled earlier in ShopifyOpen.migrate_legacy_state() before
+    Pydantic validation, which prevents hybrid state from being created.
 
     Args:
         store_ids: List of store IDs that should have state entries.
         binding: The capture binding being processed.
-        state: The current state (may be flat or dict-based).
+        state: The current state (always dict-based after ShopifyOpen migration).
         initial_state: The initial state template for new entries.
         task: The task for logging and checkpointing.
     """
-    should_checkpoint = False
-
-    # Handle legacy flat state migration
-    if isinstance(state.inc, ResourceState.Incremental):
-        # Legacy flat state - migrate to dict-based
-        if len(store_ids) != 1:
-            raise RuntimeError(
-                f"Cannot migrate legacy flat state with multiple stores ({len(store_ids)}). "
-                "Legacy captures should only have one store."
-            )
-        store_id = store_ids[0]
-        task.log.info(f"Migrating legacy flat state to dict-based for store: {store_id}")
-
-        legacy_inc = state.inc
-        state.inc = {store_id: legacy_inc}  # type: ignore[assignment]
+    # State should always be dict-based at this point (migration happens in ShopifyOpen)
+    if not isinstance(state.inc, dict) or not isinstance(initial_state.inc, dict):
+        return
 
-        if state.backfill is not None and isinstance(state.backfill, ResourceState.Backfill):
-            legacy_backfill = state.backfill
-            state.backfill = {store_id: legacy_backfill}  # type: ignore[assignment]
+    should_checkpoint = False
 
-        should_checkpoint = True
+    for store_id in store_ids:
+        inc_state_exists = store_id in state.inc
+        backfill_state_exists = (
+            isinstance(state.backfill, dict) and store_id in state.backfill
+        )
 
-    # Handle adding new stores to existing dict-based state
-    elif isinstance(state.inc, dict) and isinstance(initial_state.inc, dict):
-        for store_id in store_ids:
-            inc_state_exists = store_id in state.inc
-            backfill_state_exists = (
-                isinstance(state.backfill, dict) and store_id in state.backfill
+        if not inc_state_exists and not backfill_state_exists:
+            task.log.info(f"Initializing new state for store: {store_id}")
+            state.inc[store_id] = deepcopy(initial_state.inc[store_id])
+            if isinstance(state.backfill, dict) and isinstance(initial_state.backfill, dict):
+                state.backfill[store_id] = deepcopy(initial_state.backfill[store_id])
+            should_checkpoint = True
+        elif not inc_state_exists and backfill_state_exists:
+            # Edge case: backfill exists but incremental doesn't
+            task.log.info(
+                f"Reinitializing state for store {store_id} due to missing incremental state."
             )
-
-            if not inc_state_exists and not backfill_state_exists:
-                task.log.info(f"Initializing new state for store: {store_id}")
-                state.inc[store_id] = deepcopy(initial_state.inc[store_id])
-                if isinstance(state.backfill, dict) and isinstance(initial_state.backfill, dict):
-                    state.backfill[store_id] = deepcopy(initial_state.backfill[store_id])
-                should_checkpoint = True
-            elif not inc_state_exists and backfill_state_exists:
-                # Edge case: backfill exists but incremental doesn't
-                task.log.info(
-                    f"Reinitializing state for store {store_id} due to missing incremental state."
-                )
-                state.inc[store_id] = deepcopy(initial_state.inc[store_id])
-                if isinstance(state.backfill, dict) and isinstance(initial_state.backfill, dict):
-                    state.backfill[store_id] = deepcopy(initial_state.backfill[store_id])
-                should_checkpoint = True
+            state.inc[store_id] = deepcopy(initial_state.inc[store_id])
+            if isinstance(state.backfill, dict) and isinstance(initial_state.backfill, dict):
+                state.backfill[store_id] = deepcopy(initial_state.backfill[store_id])
+            should_checkpoint = True
 
     if should_checkpoint:
         task.log.info(f"Checkpointing reconciled state for {binding.stateKey}.")