Cypheresque - adding ability to limit output by pipeline(s)

Author: rreddy2

nodestream/cli/commands/print_schema.py

@@ -2,12 +2,13 @@
 
 from ..operations import InitializeProject, PrintProjectSchema
 from .nodestream_command import NodestreamCommand
-from .shared_options import PROJECT_FILE_OPTION
+from .shared_options import PROJECT_FILE_OPTION, MANY_PIPELINES_ARGUMENT
 
 
 class PrintSchema(NodestreamCommand):
     name = "print schema"
     description = "Print the schema for the current project"
+    arguments = [MANY_PIPELINES_ARGUMENT]
     options = [
         PROJECT_FILE_OPTION,
         option(
@@ -37,5 +38,6 @@ async def handle_async(self):
                 format_string=self.option("format"),
                 type_overrides_file=self.option("overrides"),
                 output_file=self.option("out"),
+                pipeline_names=self.argument("pipelines"),
             )
         )

nodestream/cli/operations/print_project_schema.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import Optional
+from typing import List, Optional
 
 from ...project import Project
 from ...schema.printers import SchemaPrinter
@@ -14,18 +14,26 @@ def __init__(
         format_string: str,
         type_overrides_file: Optional[str] = None,
         output_file: Optional[str] = None,
+        pipeline_names: Optional[List[str]] = None,
     ) -> None:
         self.project = project
         self.format_string = format_string
         self.output_file = output_file
         self.type_overrides_file = type_overrides_file
+        self.pipeline_names = pipeline_names or []
 
     async def perform(self, command: NodestreamCommand):
         type_overrides_file = (
             Path(self.type_overrides_file) if self.type_overrides_file else None
         )
-        schema = self.project.get_schema(type_overrides_file=type_overrides_file)
-
+        if self.pipeline_names:
+            schema = self.project.get_pipelines_schema(
+                pipeline_names=self.pipeline_names,
+                type_overrides_file=type_overrides_file
+            )
+        else:
+            schema = self.project.get_schema(type_overrides_file=type_overrides_file)
+        
         # Import all schema printers so that they can register themselves
         SchemaPrinter.import_all()
         printer = SchemaPrinter.from_name(self.format_string)

nodestream/project/project.py

@@ -291,6 +291,61 @@ def get_schema(self, type_overrides_file: Optional[Path] = None) -> Schema:
             schema.merge(overrides_schema)
         return schema
 
+    def get_pipelines_schema(self, pipeline_names: List[str], type_overrides_file: Optional[Path] = None) -> Schema:
+        """Returns a `GraphSchema` representing only the specified pipelines.
+
+        This method generates a schema from only the specified pipelines,
+        allowing you to see what schema specific pipelines contribute without interference
+        from other pipelines in the project.
+
+        Args:
+            pipeline_names (List[str]): List of pipeline names to include in schema generation.
+            type_overrides_file (Optional[Path], optional): A path to a YAML file containing type overrides. Defaults to None.
+
+        Returns:
+            Schema: The schema representing only the specified pipelines.
+
+        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)
+        
+        if not pipelines_found:
+            available_pipelines = [
+                name for scope in self.scopes_by_name.values() 
+                for name in scope.pipelines_by_name.keys()
+            ]
+            raise ValueError(f"None of the specified pipelines {pipeline_names} were found. Available pipelines: {available_pipelines}")
+        
+        schema = filtered_project.make_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()