source-shopify-native: simplify state handling to follow stripe pattern

Remove complex state migration logic (ShopifyRequest, ShopifyOpen) and follow
the pattern from source-stripe-native instead:

- New captures: always use dict-based state {"inc": {"store_id": {...}}}
- Legacy single-store captures: continue with flat state {"inc": {"cursor": ...}}
- Transitioning to multiple stores: requires backfill (already enforced)

This avoids hybrid state issues by never changing the state format at runtime.
Legacy captures keep their flat state, new captures start with dict-based state.

🤖 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,5 +1,5 @@
 from logging import Logger
-from typing import Callable, Awaitable, Union
+from typing import Callable, Awaitable
 
 from estuary_cdk.flow import (
     ConnectorSpec,
@@ -14,26 +14,16 @@
 )
 from estuary_cdk.capture.common import ResourceConfig
 
+from estuary_cdk.capture import Request
+
 from .resources import all_resources, validate_credentials
 from .models import (
     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:
@@ -170,7 +160,7 @@ class Connector(
     BaseCaptureConnector[EndpointConfig, ResourceConfig, ConnectorState],
 ):
     def request_class(self):
-        return ShopifyRequest
+        return Request[EndpointConfig, ResourceConfig, ConnectorState]
 
     async def spec(self, log: Logger, _: request.Spec) -> ConnectorSpec:
         return ConnectorSpec(

source-shopify-native/source_shopify_native/models.py

@@ -556,118 +556,5 @@ 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

@@ -138,18 +138,39 @@ def _create_initial_state(
     store_ids: list[str],
     start_date: AwareDatetime,
     use_backfill: bool,
+    use_multi_store_keys: bool,
 ) -> ResourceState:
-    """Create initial state for a resource. State is always dict-based, keyed by store_id."""
+    """Create initial state for a resource.
+
+    State format depends on use_multi_store_keys:
+    - False (1 store): Flat state format {"inc": {"cursor": "..."}}
+    - True (multiple stores): Dict-based {"inc": {"store_id": {"cursor": "..."}}}
+
+    This follows the same pattern as source-stripe-native: flat state for single account,
+    dict-based for multiple accounts. Transitioning requires backfill.
+    """
     cutoff = datetime.now(tz=UTC)
 
-    if use_backfill:
+    if use_multi_store_keys:
+        # Dict-based state for multiple stores
+        if use_backfill:
+            return ResourceState(
+                inc={sid: ResourceState.Incremental(cursor=cutoff) for sid in store_ids},  # type: ignore[arg-type]
+                backfill={sid: ResourceState.Backfill(next_page=dt_to_str(start_date), cutoff=cutoff) for sid in store_ids},  # type: ignore[arg-type]
+            )
         return ResourceState(
-            inc={sid: ResourceState.Incremental(cursor=cutoff) for sid in store_ids},  # type: ignore[arg-type]
-            backfill={sid: ResourceState.Backfill(next_page=dt_to_str(start_date), cutoff=cutoff) for sid in store_ids},  # type: ignore[arg-type]
+            inc={sid: ResourceState.Incremental(cursor=start_date) for sid in store_ids},  # type: ignore[arg-type]
+        )
+    else:
+        # Flat state for single store (backward compatible with legacy)
+        if use_backfill:
+            return ResourceState(
+                inc=ResourceState.Incremental(cursor=cutoff),
+                backfill=ResourceState.Backfill(next_page=dt_to_str(start_date), cutoff=cutoff),
+            )
+        return ResourceState(
+            inc=ResourceState.Incremental(cursor=start_date),
         )
-    return ResourceState(
-        inc={sid: ResourceState.Incremental(cursor=start_date) for sid in store_ids},  # type: ignore[arg-type]
-    )
 
 
 def _reconcile_connector_state(
@@ -161,48 +182,53 @@ def _reconcile_connector_state(
 ) -> None:
     """Reconcile connector state to ensure all stores have proper state entries.
 
-    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.
+    This follows the pattern from source-stripe-native: only add new stores to
+    existing dict-based state. Flat state (single store) is left unchanged.
+
+    State format is determined by _create_initial_state based on use_multi_store_keys:
+    - Single store: flat state, no reconciliation needed
+    - Multiple stores: dict-based state, add new stores as needed
 
     Args:
         store_ids: List of store IDs that should have state entries.
         binding: The capture binding being processed.
-        state: The current state (always dict-based after ShopifyOpen migration).
+        state: The current state.
         initial_state: The initial state template for new entries.
         task: The task for logging and checkpointing.
     """
-    # 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
-
-    should_checkpoint = False
-
-    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."
+    # Only reconcile dict-based state (multiple stores)
+    # Flat state (single store) doesn't need reconciliation
+    if (
+        isinstance(state.inc, dict)
+        and isinstance(initial_state.inc, dict)
+    ):
+        should_checkpoint = False
+
+        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
             )
-            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}.")
-        task.checkpoint(ConnectorState(bindingStateV1={binding.stateKey: 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
+
+        if should_checkpoint:
+            task.log.info(f"Checkpointing reconciled state for {binding.stateKey}.")
+            task.checkpoint(ConnectorState(bindingStateV1={binding.stateKey: state}))
 
 
 async def _check_plan_allows_pii(http: HTTPMixin, url: str, log: Logger) -> bool:
@@ -277,15 +303,19 @@ async def all_resources(
 ) -> list[Resource]:
     """Discover all available resources across all configured stores.
 
-    State is always dict-based internally, keyed by store_id.
-    Collection keys depend on use_multi_store_keys:
-    - True: ["/_meta/store", "/id"] (new captures and multi-store)
-    - False: ["/id"] (legacy single-store captures for backward compatibility)
+    State and collection key format depends on use_multi_store_keys:
+    - True: Dict-based state {"inc": {"store_id": {...}}}, keys ["/_meta/store", "/id"]
+    - False: Flat state {"inc": {"cursor": "..."}}, keys ["/id"]
+
+    This follows the pattern from source-stripe-native: new captures always use
+    dict-based state, legacy single-store captures continue with flat state,
+    and transitioning requires backfill.
 
     Args:
-        use_multi_store_keys: Whether to include /_meta/store in collection keys.
-            New captures should always pass True. Only pass False for legacy
-            single-store captures that haven't added additional stores.
+        use_multi_store_keys: Whether to use multi-store format.
+            True: New captures, existing multi-store captures, or legacy captures
+                  transitioning to multiple stores (after backfill acknowledgment).
+            False: Legacy single-store captures (backward compatibility).
     """
     # Build store contexts
     store_contexts: dict[str, dict] = {}
@@ -330,7 +360,7 @@ async def all_resources(
             continue
 
         use_backfill = not model.SHOULD_USE_BULK_QUERIES and model.SORT_KEY is not None
-        initial_state = _create_initial_state(stores_with_access, config.start_date, use_backfill)
+        initial_state = _create_initial_state(stores_with_access, config.start_date, use_backfill, use_multi_store_keys)
 
         def create_open_fn(
             model: type[ShopifyGraphQLResource],