Strict typing for Pipeline object

Summary: The `Pipeline` contains a set of `PipelineStep`s, each of which converts an input type to an output type. Typically, both the input and output types are `TraceGraph` so it has been safe to concatenate steps without type checking.

Reviewed By: fahndrich

Differential Revision: D66710498

fbshipit-source-id: 362c0c294db4327a98d323db2e702d7d635ef3fe

sapp/cli_lib.py

@@ -24,7 +24,7 @@
 from .filesystem import find_root
 from .json_cmd import json_cmd
 from .models import PrimaryKeyGenerator, Run
-from .pipeline import Pipeline, Summary
+from .pipeline import PipelineBuilder, Summary
 from .pipeline.add_features import AddFeatures
 from .pipeline.create_database import CreateDatabase
 from .pipeline.database_saver import DatabaseSaver
@@ -192,19 +192,16 @@ def analyze(
     else:
         analysis_output = AnalysisOutput.from_file(input_file)
 
-    pipeline_steps = [
-        ctx.parser_class(),
-        CreateDatabase(ctx.database),
-        AddFeatures(add_feature),
-        ModelGenerator(),
-        TrimTraceGraph(),
-        DatabaseSaver(ctx.database, Run, PrimaryKeyGenerator(), dry_run),
-    ]
-    # pyre-fixme[6]: Expected
-    #  `List[tools.sapp.sapp.pipeline.PipelineStep[typing.Any, typing.Any]]` for 1st
-    #  param but got `List[typing.Union[DatabaseSaver, ModelGenerator, TrimTraceGraph,
-    #  tools.sapp.sapp.base_parser.BaseParser]]`.
-    pipeline = Pipeline(pipeline_steps)
+    pipeline = (
+        PipelineBuilder()
+        .append(ctx.parser_class())
+        .append(CreateDatabase(ctx.database))
+        .append(AddFeatures(add_feature))
+        .append(ModelGenerator())
+        .append(TrimTraceGraph())
+        .append(DatabaseSaver(ctx.database, Run, PrimaryKeyGenerator(), dry_run))
+        .build()
+    )
     pipeline.run(analysis_output, summary_blob)
 
 

sapp/pipeline/__init__.py

@@ -365,3 +365,16 @@ def run(
             ", ".join([f"{name} took {time_str(delta)}" for name, delta in timing]),
         )
         return next_input, summary
+
+
+class PipelineBuilder(Generic[T_in]):
+    def __init__(self) -> None:
+        # pyre-fixme[4]: Attribute annotation cannot contain `Any`.
+        self.steps: List[PipelineStep[Any, Any]] = []
+
+    def append(self, step: PipelineStep[T_in, T_out]) -> "PipelineBuilder[T_out]":
+        self.steps.append(step)
+        return cast(PipelineBuilder[T_out], self)
+
+    def build(self) -> Pipeline:
+        return Pipeline(self.steps)