From 44992d3447efdcb8620280ff9fbb4457b3a00523 Mon Sep 17 00:00:00 2001 From: Serhii Lazebnyi Date: Sat, 1 Aug 2026 00:02:49 +0100 Subject: [PATCH] Remove unused SubstreamPartitionRouter._migrate_child_state_to_parent_state --- .../substream_partition_router.py | 62 ------------------- 1 file changed, 62 deletions(-) diff --git a/airbyte_cdk/sources/declarative/partition_routers/substream_partition_router.py b/airbyte_cdk/sources/declarative/partition_routers/substream_partition_router.py index 5d2524b55b..490182f37b 100644 --- a/airbyte_cdk/sources/declarative/partition_routers/substream_partition_router.py +++ b/airbyte_cdk/sources/declarative/partition_routers/substream_partition_router.py @@ -318,68 +318,6 @@ def _extract_extra_fields( extracted_extra_fields[".".join(extra_field_path)] = extra_field_value return extracted_extra_fields - def _migrate_child_state_to_parent_state(self, stream_state: StreamState) -> StreamState: - """ - Migrate the child or global stream state into the parent stream's state format. - - This method converts the child stream state—or, if present, the global state—into a format that is - compatible with parent streams that use incremental synchronization. The migration occurs only for - parent streams with incremental dependencies. It filters out per-partition states and retains only the - global state in the form {cursor_field: cursor_value}. - - The method supports multiple input formats: - - A simple global state, e.g.: - {"updated_at": "2023-05-27T00:00:00Z"} - - A state object that contains a "state" key (which is assumed to hold the global state), e.g.: - {"state": {"updated_at": "2023-05-27T00:00:00Z"}, ...} - In this case, the migration uses the first value from the "state" dictionary. - - Any per-partition state formats or other non-simple structures are ignored during migration. - - Args: - stream_state (StreamState): The state to migrate. Expected formats include: - - {"updated_at": "2023-05-27T00:00:00Z"} - - {"state": {"updated_at": "2023-05-27T00:00:00Z"}, ...} - (In this format, only the first global state value is used, and per-partition states are ignored.) - - Returns: - StreamState: A migrated state for parent streams in the format: - { - "parent_stream_name": {"parent_stream_cursor": "2023-05-27T00:00:00Z"} - } - where each parent stream with an incremental dependency is assigned its corresponding cursor value. - - Example: - Input: {"updated_at": "2023-05-27T00:00:00Z"} - Output: { - "parent_stream_name": {"parent_stream_cursor": "2023-05-27T00:00:00Z"} - } - """ - substream_state_values = list(stream_state.values()) - substream_state = substream_state_values[0] if substream_state_values else {} - - # Ignore per-partition states or invalid formats. - if isinstance(substream_state, (list, dict)) or len(substream_state_values) != 1: - # If a global state is present under the key "state", use its first value. - if ( - "state" in stream_state - and isinstance(stream_state["state"], dict) - and stream_state["state"] != {} - ): - substream_state = list(stream_state["state"].values())[0] - else: - return {} - - # Build the parent state for all parent streams with incremental dependencies. - parent_state = {} - if substream_state: - for parent_config in self.parent_stream_configs: - if parent_config.incremental_dependency: - parent_state[parent_config.stream.name] = { - parent_config.stream.cursor_field: substream_state - } - - return parent_state - def get_stream_state(self) -> Optional[Mapping[str, StreamState]]: """ Get the state of the parent streams.