refactor: move stream_name_to_group into ModelToComponentFactory

- Add stream_name_to_group parameter to ModelToComponentFactory.__init__()
- Add set_stream_name_to_group() method for post-init configuration
- Factory now looks up block_simultaneous_read from its own mapping
- Remove config injection hack from ConcurrentDeclarativeSource.streams()
- Update tests to use factory-based approach instead of extra fields

Co-Authored-By: unknown <>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -405,13 +405,11 @@ 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)
 
-        stream_name_to_group = self._build_stream_name_to_group(self._source_config)
+        self._constructor.set_stream_name_to_group(
+            self._build_stream_name_to_group(self._source_config)
+        )
 
         prepared_configs = self._initialize_cache_for_parent_streams(deepcopy(stream_configs))
-        for stream_config in prepared_configs:
-            stream_name = stream_config.get("name", "")
-            if stream_name in stream_name_to_group:
-                stream_config["block_simultaneous_read"] = stream_name_to_group[stream_name]
 
         source_streams = [
             self._constructor.create_component(

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -682,6 +682,7 @@ 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
@@ -698,9 +699,14 @@ 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 {}
         # 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 _init_mappings(self) -> None:
         self.PYDANTIC_MODEL_TO_CONSTRUCTOR: Mapping[Type[BaseModel], Callable[..., Any]] = {
             AddedFieldDefinitionModel: self.create_added_field_definition,
@@ -2118,7 +2124,7 @@ def create_default_stream(
             logger=logging.getLogger(f"airbyte.{stream_name}"),
             cursor=concurrent_cursor,
             supports_file_transfer=hasattr(model, "file_uploader") and bool(model.file_uploader),
-            block_simultaneous_read=getattr(model, "block_simultaneous_read", "") or "",
+            block_simultaneous_read=self._stream_name_to_group.get(stream_name, ""),
         )
 
     def _migrate_state(self, model: DeclarativeStreamModel, config: Config) -> None:

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -5215,19 +5215,17 @@ def test_catalog_defined_cursor_field_stream_missing():
 
 
 def test_block_simultaneous_read_from_stream_groups():
-    """Test that block_simultaneous_read flows through from stream_groups to DefaultStream.
+    """Test that block_simultaneous_read flows through from stream_name_to_group to DefaultStream.
 
-    The stream_groups config is processed by ConcurrentDeclarativeSource which injects
-    block_simultaneous_read into individual stream configs before passing them to the factory.
-    This test verifies that the factory correctly reads block_simultaneous_read from the
-    extra fields on the stream config dict.
+    The stream_groups config is parsed by ConcurrentDeclarativeSource into a stream_name_to_group
+    mapping, which is then set on the ModelToComponentFactory. The factory uses this mapping to
+    look up the group for each stream it creates.
     """
     content = """
     parent_stream:
       type: DeclarativeStream
       name: "parent"
       primary_key: "id"
-      block_simultaneous_read: "issues_endpoint"
       retriever:
         type: SimpleRetriever
         requester:
@@ -5255,7 +5253,6 @@ def test_block_simultaneous_read_from_stream_groups():
       type: DeclarativeStream
       name: "child"
       primary_key: "id"
-      block_simultaneous_read: "issues_endpoint"
       retriever:
         type: SimpleRetriever
         requester:
@@ -5318,26 +5315,31 @@ def test_block_simultaneous_read_from_stream_groups():
 
     config = {"api_key": "test_key"}
 
+    # Create a factory with stream_name_to_group mapping (as ConcurrentDeclarativeSource would do)
+    factory_with_groups = ModelToComponentFactory(
+        stream_name_to_group={"parent": "issues_endpoint", "child": "issues_endpoint"}
+    )
+
     parsed_manifest = YamlDeclarativeSource._parse(content)
     resolved_manifest = resolver.preprocess_manifest(parsed_manifest)
 
-    # Test parent stream with block_simultaneous_read injected (as ConcurrentDeclarativeSource would do)
+    # Test parent stream gets block_simultaneous_read from the factory's stream_name_to_group
     parent_manifest = transformer.propagate_types_and_parameters(
         "", resolved_manifest["parent_stream"], {}
     )
-    parent_stream: DefaultStream = factory.create_component(
+    parent_stream: DefaultStream = factory_with_groups.create_component(
         model_type=DeclarativeStreamModel, component_definition=parent_manifest, config=config
     )
 
     assert isinstance(parent_stream, DefaultStream)
     assert parent_stream.name == "parent"
     assert parent_stream.block_simultaneous_read == "issues_endpoint"
 
-    # Test child stream with block_simultaneous_read injected
+    # Test child stream gets block_simultaneous_read from the factory's stream_name_to_group
     child_manifest = transformer.propagate_types_and_parameters(
         "", resolved_manifest["child_stream"], {}
     )
-    child_stream: DefaultStream = factory.create_component(
+    child_stream: DefaultStream = factory_with_groups.create_component(
         model_type=DeclarativeStreamModel, component_definition=child_manifest, config=config
     )
 
@@ -5349,7 +5351,7 @@ def test_block_simultaneous_read_from_stream_groups():
     no_block_manifest = transformer.propagate_types_and_parameters(
         "", resolved_manifest["no_block_stream"], {}
     )
-    no_block_stream: DefaultStream = factory.create_component(
+    no_block_stream: DefaultStream = factory_with_groups.create_component(
         model_type=DeclarativeStreamModel, component_definition=no_block_manifest, config=config
     )
 
@@ -5358,6 +5360,62 @@ def test_block_simultaneous_read_from_stream_groups():
     assert no_block_stream.block_simultaneous_read == ""
 
 
+def test_set_stream_name_to_group():
+    """Test that set_stream_name_to_group updates the factory's stream_name_to_group mapping."""
+    content = """
+    test_stream:
+      type: DeclarativeStream
+      name: "test"
+      primary_key: "id"
+      retriever:
+        type: SimpleRetriever
+        requester:
+          type: HttpRequester
+          url_base: "https://api.example.com"
+          path: "/test"
+          http_method: "GET"
+          authenticator:
+            type: BearerAuthenticator
+            api_token: "{{ config['api_key'] }}"
+        record_selector:
+          type: RecordSelector
+          extractor:
+            type: DpathExtractor
+            field_path: []
+      schema_loader:
+        type: InlineSchemaLoader
+        schema:
+          type: object
+          properties:
+            id:
+              type: string
+    """
+
+    config = {"api_key": "test_key"}
+
+    # Create factory without stream_name_to_group
+    test_factory = ModelToComponentFactory()
+
+    parsed_manifest = YamlDeclarativeSource._parse(content)
+    resolved_manifest = resolver.preprocess_manifest(parsed_manifest)
+    stream_manifest = transformer.propagate_types_and_parameters(
+        "", resolved_manifest["test_stream"], {}
+    )
+
+    # Without stream_name_to_group, block_simultaneous_read should be empty
+    stream: DefaultStream = test_factory.create_component(
+        model_type=DeclarativeStreamModel, component_definition=stream_manifest, config=config
+    )
+    assert stream.block_simultaneous_read == ""
+
+    # After setting stream_name_to_group, block_simultaneous_read should be populated
+    test_factory.set_stream_name_to_group({"test": "my_group"})
+    stream = test_factory.create_component(
+        model_type=DeclarativeStreamModel, component_definition=stream_manifest, config=config
+    )
+    assert stream.block_simultaneous_read == "my_group"
+
+
 def get_schema_loader(stream: DefaultStream):
     assert isinstance(
         stream._stream_partition_generator._partition_factory._schema_loader,