refactor: move _build_stream_name_to_group into ModelToComponentFactory

- Factory now owns the stream_groups resolution via set_stream_groups(manifest)
- ConcurrentDeclarativeSource just calls factory.set_stream_groups(manifest)
- Removed _build_stream_name_to_group from ConcurrentDeclarativeSource
- Updated tests to use factory's _build_stream_name_to_group directly

Co-Authored-By: unknown <>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -405,9 +405,7 @@ def streams(self, config: Mapping[str, Any]) -> List[AbstractStream]:  # type: i
         if api_budget_model:
             self._constructor.set_api_budget(api_budget_model, self._config)
 
-        self._constructor.set_stream_name_to_group(
-            self._build_stream_name_to_group(self._source_config)
-        )
+        self._constructor.set_stream_groups(self._source_config)
 
         prepared_configs = self._initialize_cache_for_parent_streams(deepcopy(stream_configs))
 
@@ -532,38 +530,6 @@ def dynamic_streams(self) -> List[Dict[str, Any]]:
             with_dynamic_stream_name=True,
         )
 
-    @staticmethod
-    def _build_stream_name_to_group(manifest: Mapping[str, Any]) -> Dict[str, str]:
-        """Build a mapping from stream name to group name based on the stream_groups manifest config.
-
-        After manifest reference resolution, each stream reference in stream_groups.streams
-        is resolved to the full stream definition dict containing a 'name' field.
-
-        Returns:
-            A dict mapping stream name -> group name for streams that belong to a group.
-        """
-        stream_name_to_group: Dict[str, str] = {}
-        stream_groups = manifest.get("stream_groups", {})
-        if not stream_groups:
-            return stream_name_to_group
-
-        for group_name, group_config in stream_groups.items():
-            streams = group_config.get("streams", [])
-            for stream_ref in streams:
-                if isinstance(stream_ref, dict):
-                    # After reference resolution, stream_ref is a full stream definition dict
-                    stream_name = stream_ref.get("name", "")
-                    if stream_name:
-                        stream_name_to_group[stream_name] = group_name
-                elif isinstance(stream_ref, str):
-                    # If not resolved (shouldn't happen normally), extract name from ref path
-                    # e.g., "#/definitions/my_stream" -> "my_stream"
-                    if stream_ref.startswith("#/definitions/"):
-                        stream_name = stream_ref.split("/")[-1]
-                        stream_name_to_group[stream_name] = group_name
-
-        return stream_name_to_group
-
     def _stream_configs(self, manifest: Mapping[str, Any]) -> List[Dict[str, Any]]:
         # This has a warning flag for static, but after we finish part 4 we'll replace manifest with self._source_config
         stream_configs = []

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -682,7 +682,6 @@ def __init__(
         max_concurrent_async_job_count: Optional[int] = None,
         configured_catalog: Optional[ConfiguredAirbyteCatalog] = None,
         api_budget: Optional[APIBudget] = None,
-        stream_name_to_group: Optional[Dict[str, str]] = None,
     ):
         self._init_mappings()
         self._limit_pages_fetched_per_slice = limit_pages_fetched_per_slice
@@ -699,13 +698,49 @@ def __init__(
         self._connector_state_manager = connector_state_manager or ConnectorStateManager()
         self._api_budget: Optional[Union[APIBudget]] = api_budget
         self._job_tracker: JobTracker = JobTracker(max_concurrent_async_job_count or 1)
-        self._stream_name_to_group: Dict[str, str] = stream_name_to_group or {}
+        self._stream_name_to_group: Dict[str, str] = {}
         # placeholder for deprecation warnings
         self._collected_deprecation_logs: List[ConnectorBuilderLogMessage] = []
 
-    def set_stream_name_to_group(self, stream_name_to_group: Dict[str, str]) -> None:
-        """Set the mapping from stream name to group name for block_simultaneous_read."""
-        self._stream_name_to_group = stream_name_to_group
+    def set_stream_groups(self, manifest: Mapping[str, Any]) -> None:
+        """Build and set the stream-name-to-group mapping from the manifest's stream_groups config.
+
+        After manifest reference resolution, each stream reference in stream_groups.streams
+        is resolved to the full stream definition dict containing a 'name' field.
+        """
+        self._stream_name_to_group = self._build_stream_name_to_group(manifest)
+
+    @staticmethod
+    def _build_stream_name_to_group(manifest: Mapping[str, Any]) -> Dict[str, str]:
+        """Build a mapping from stream name to group name based on the stream_groups manifest config.
+
+        After manifest reference resolution, each stream reference in stream_groups.streams
+        is resolved to the full stream definition dict containing a 'name' field.
+
+        Returns:
+            A dict mapping stream name -> group name for streams that belong to a group.
+        """
+        stream_name_to_group: Dict[str, str] = {}
+        stream_groups = manifest.get("stream_groups", {})
+        if not stream_groups:
+            return stream_name_to_group
+
+        for group_name, group_config in stream_groups.items():
+            streams = group_config.get("streams", [])
+            for stream_ref in streams:
+                if isinstance(stream_ref, dict):
+                    # After reference resolution, stream_ref is a full stream definition dict
+                    stream_name = stream_ref.get("name", "")
+                    if stream_name:
+                        stream_name_to_group[stream_name] = group_name
+                elif isinstance(stream_ref, str):
+                    # If not resolved (shouldn't happen normally), extract name from ref path
+                    # e.g., "#/definitions/my_stream" -> "my_stream"
+                    if stream_ref.startswith("#/definitions/"):
+                        stream_name = stream_ref.split("/")[-1]
+                        stream_name_to_group[stream_name] = group_name
+
+        return stream_name_to_group
 
     def _init_mappings(self) -> None:
         self.PYDANTIC_MODEL_TO_CONSTRUCTOR: Mapping[Type[BaseModel], Callable[..., Any]] = {

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -45,9 +45,6 @@
 from airbyte_cdk.sources.declarative.auth.token_provider import SessionTokenProvider
 from airbyte_cdk.sources.declarative.checks import CheckStream
 from airbyte_cdk.sources.declarative.concurrency_level import ConcurrencyLevel
-from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
-    ConcurrentDeclarativeSource,
-)
 from airbyte_cdk.sources.declarative.datetime.min_max_datetime import MinMaxDatetime
 from airbyte_cdk.sources.declarative.decoders import JsonDecoder, PaginationDecoderDecorator
 from airbyte_cdk.sources.declarative.extractors import DpathExtractor, RecordFilter, RecordSelector
@@ -5323,11 +5320,9 @@ def test_block_simultaneous_read_from_stream_groups():
     parsed_manifest = YamlDeclarativeSource._parse(content)
     resolved_manifest = resolver.preprocess_manifest(parsed_manifest)
 
-    # Build stream_name_to_group from the manifest's stream_groups (as ConcurrentDeclarativeSource does)
-    stream_name_to_group = ConcurrentDeclarativeSource._build_stream_name_to_group(
-        resolved_manifest
-    )
-    factory_with_groups = ModelToComponentFactory(stream_name_to_group=stream_name_to_group)
+    # Use the factory's set_stream_groups to resolve stream_groups from the manifest
+    factory_with_groups = ModelToComponentFactory()
+    factory_with_groups.set_stream_groups(resolved_manifest)
 
     # Test parent stream gets block_simultaneous_read from stream_groups
     parent_manifest = transformer.propagate_types_and_parameters(

unit_tests/sources/declarative/test_concurrent_declarative_source.py

@@ -56,6 +56,9 @@
 from airbyte_cdk.sources.declarative.extractors.record_filter import (
     ClientSideIncrementalRecordFilterDecorator,
 )
+from airbyte_cdk.sources.declarative.parsers.model_to_component_factory import (
+    ModelToComponentFactory,
+)
 from airbyte_cdk.sources.declarative.partition_routers import AsyncJobPartitionRouter
 from airbyte_cdk.sources.declarative.retrievers.simple_retriever import SimpleRetriever
 from airbyte_cdk.sources.declarative.stream_slicers.declarative_partition_generator import (
@@ -5219,5 +5222,5 @@ def test_given_record_selector_is_filtering_when_read_then_raise_error():
 )
 def test_build_stream_name_to_group(manifest, expected):
     """Test _build_stream_name_to_group correctly maps stream names to group names."""
-    result = ConcurrentDeclarativeSource._build_stream_name_to_group(manifest)
+    result = ModelToComponentFactory._build_stream_name_to_group(manifest)
     assert result == expected