updating projects to be simpler

Author: rreddy2

nodestream/project/project.py

@@ -13,7 +13,12 @@
 from ..pipeline import Step
 from ..pipeline.object_storage import ObjectStore
 from ..pluggable import Pluggable
-from ..schema import ExpandsSchema, ExpandsSchemaFromChildren, Schema
+from ..schema import (
+    ExpandsSchema,
+    ExpandsSchemaFromChildren,
+    Schema,
+    SchemaExpansionCoordinator,
+)
 from .pipeline_definition import PipelineDefinition
 from .pipeline_scope import PipelineScope
 from .plugin import PluginConfiguration
@@ -245,6 +250,16 @@ def get_pipeline_by_name(self, pipeline_name: str) -> PipelineDefinition:
             if pipeline is not None:
                 return pipeline
 
+    def get_pipeline_by_names(
+        self, pipeline_names: List[str]
+    ) -> Iterable[PipelineDefinition]:
+        """Returns pipeline objects in the project."""
+        for scope in self.scopes_by_name.values():
+            for pipeline_name in pipeline_names:
+                pipeline = scope.pipelines_by_name.get(pipeline_name, None)
+                if pipeline is not None:
+                    yield pipeline
+
     def delete_pipeline(
         self,
         scope_name: Optional[str],
@@ -310,32 +325,9 @@ def get_pipelines_schema(
         Raises:
             ValueError: If none of the specified pipelines are found.
         """
-        # Create a temporary project to store only the specified pipelines
-        filtered_project = Project(
-            targets_by_name=self.targets_by_name,
-            storage_configuration=self.storage_configuration,
-        )
-
-        pipelines_found = []
-
-        for scope in self.scopes_by_name.values():
-            filtered_scope = PipelineScope(
-                name=scope.name,
-                config=scope.config,
-                pipeline_configuration=scope.pipeline_configuration,
-            )
-
-            for pipeline_name in pipeline_names:
-                if pipeline_name in scope.pipelines_by_name:
-                    filtered_scope.add_pipeline_definition(
-                        scope.pipelines_by_name[pipeline_name]
-                    )
-                    pipelines_found.append(pipeline_name)
-
-            if filtered_scope.pipelines_by_name:
-                filtered_project.add_scope(filtered_scope)
+        pipeline_definitions = list(self.get_pipeline_by_names(pipeline_names))
 
-        if not pipelines_found:
+        if not pipeline_definitions:
             available_pipelines = [
                 name
                 for scope in self.scopes_by_name.values()
@@ -345,7 +337,19 @@ def get_pipelines_schema(
                 f"None of the specified pipelines {pipeline_names} were found. Available pipelines: {available_pipelines}"
             )
 
-        return filtered_project.get_schema(type_overrides_file=type_overrides_file)
+        coordinator = SchemaExpansionCoordinator(schema := Schema())
+
+        for pipeline in pipeline_definitions:
+            pipeline.initialize_for_introspection()
+            pipeline.expand_schema(coordinator=coordinator)
+
+        schema = coordinator.schema
+
+        if type_overrides_file is not None:
+            overrides_schema = Schema.read_from_file(type_overrides_file)
+            schema.merge(overrides_schema)
+
+        return schema
 
     def get_child_expanders(self) -> Iterable[ExpandsSchema]:
         return self.scopes_by_name.values()

tests/unit/project/test_project.py

@@ -379,73 +379,79 @@ def test_project_get_from_storage(project, mocker):
         same_instance(project.storage_configuration.initialize_by_name.return_value),
     )
 
-
-def test_get_pipelines_schema_single_pipeline(project, mocker):
-    """Test filtering schema for a single pipeline."""
+def test_get_pipelines_schema_multiple_pipelines(project, mocker):
     mock_schema = mocker.Mock(spec=Schema)
-
-    # Mock the pipeline initialization to avoid file system dependencies
-    mocker.patch.object(project.__class__, "make_schema", return_value=mock_schema)
-    mock_pipeline_init = mocker.patch(
-        "nodestream.project.pipeline_definition.PipelineDefinition.initialize_for_introspection"
-    )
-    mock_pipeline = mocker.Mock()
-    mock_pipeline.expand_schema = mocker.Mock()
-    mock_pipeline_init.return_value = mock_pipeline
-
-    result = project.get_pipelines_schema(["test"])
-
-    # Verify the result
+    mock_coordinator = mocker.Mock()
+    mock_coordinator.schema = mock_schema
+
+    mocker.patch('nodestream.project.project.SchemaExpansionCoordinator', return_value=mock_coordinator)
+    
+    mock_pipeline1 = mocker.Mock()
+    mock_pipeline1.initialize_for_introspection = mocker.Mock()
+    mock_pipeline1.expand_schema = mocker.Mock()
+    
+    mock_pipeline2 = mocker.Mock()
+    mock_pipeline2.initialize_for_introspection = mocker.Mock()
+    mock_pipeline2.expand_schema = mocker.Mock()
+    
+    mocker.patch.object(project, 'get_pipeline_by_names', return_value=[mock_pipeline1, mock_pipeline2])
+    
+    result = project.get_pipelines_schema(["test", "test2"])
+    
+    mock_pipeline1.initialize_for_introspection.assert_called_once()
+    mock_pipeline1.expand_schema.assert_called_once_with(coordinator=mock_coordinator)
+    mock_pipeline2.initialize_for_introspection.assert_called_once()
+    mock_pipeline2.expand_schema.assert_called_once_with(coordinator=mock_coordinator)
     assert_that(result, same_instance(mock_schema))
 
 
 def test_get_pipelines_schema_with_type_overrides(project, mocker):
-    """Test pipeline schema generation with type overrides."""
     mock_base_schema = mocker.Mock(spec=Schema)
     mock_overrides_schema = mocker.Mock(spec=Schema)
-
-    # Mock the pipeline initialization and schema loading
-    mocker.patch.object(project.__class__, "make_schema", return_value=mock_base_schema)
-    mock_pipeline_init = mocker.patch(
-        "nodestream.project.pipeline_definition.PipelineDefinition.initialize_for_introspection"
-    )
+    mock_coordinator = mocker.Mock()
+    mock_coordinator.schema = mock_base_schema
+    
+    mocker.patch('nodestream.project.project.SchemaExpansionCoordinator', return_value=mock_coordinator)
+    Schema.read_from_file = mocker.Mock(return_value=mock_overrides_schema)
+    
     mock_pipeline = mocker.Mock()
+    mock_pipeline.initialize_for_introspection = mocker.Mock()
     mock_pipeline.expand_schema = mocker.Mock()
-    mock_pipeline_init.return_value = mock_pipeline
-
-    Schema.read_from_file = mocker.Mock(return_value=mock_overrides_schema)
-
+    mocker.patch.object(project, 'get_pipeline_by_names', return_value=[mock_pipeline])
+    
     overrides_path = Path("some/overrides.yaml")
     result = project.get_pipelines_schema(["test"], overrides_path)
-
+    
     Schema.read_from_file.assert_called_once_with(overrides_path)
     mock_base_schema.merge.assert_called_once_with(mock_overrides_schema)
     assert_that(result, same_instance(mock_base_schema))
 
 
-def test_get_pipelines_schema_nonexistent_pipeline_raises_error(project):
-    """Test that specifying non-existent pipelines raises ValueError with helpful message."""
+def test_get_pipelines_schema_nonexistent_pipeline_raises_error(project, mocker):
+    mocker.patch.object(project, 'get_pipeline_by_names', return_value=[])
+    
     with pytest.raises(ValueError) as exc_info:
         project.get_pipelines_schema(["nonexistent"])
-
+    
     error_message = str(exc_info.value)
-    assert_that(
-        error_message,
-        equal_to(
-            "None of the specified pipelines ['nonexistent'] were found. Available pipelines: ['test', 'test2']"
-        ),
-    )
+    assert_that(error_message, equal_to(
+        "None of the specified pipelines ['nonexistent'] were found. Available pipelines: ['test', 'test2']"
+    ))
 
+def test_get_pipeline_by_names(project):
+    pipelines = list(project.get_pipeline_by_names(["test", "test2"]))
+    
+    assert_that(pipelines, has_length(2))
+    assert_that([p.name for p in pipelines], contains_inanyorder("test", "test2"))
 
-def test_get_pipelines_schema_empty_list_raises_error(project):
-    """Test that providing an empty pipeline list raises ValueError."""
-    with pytest.raises(ValueError) as exc_info:
-        project.get_pipelines_schema([])
 
-    error_message = str(exc_info.value)
-    assert_that(
-        error_message,
-        equal_to(
-            "None of the specified pipelines [] were found. Available pipelines: ['test', 'test2']"
-        ),
-    )
+def test_get_pipeline_by_names_nonexistent_pipeline(project):
+    pipelines = list(project.get_pipeline_by_names(["nonexistent"]))
+    assert_that(pipelines, has_length(0))
+
+
+def test_get_pipeline_by_names_mixed_existing_nonexisting(project):
+    pipelines = list(project.get_pipeline_by_names(["test", "nonexistent", "test2"]))
+    
+    assert_that(pipelines, has_length(2))
+    assert_that([p.name for p in pipelines], contains_inanyorder("test", "test2"))