python_modules/dagster/dagster/core/execution_plan/plan.py
def build(pipeline_def, environment_config):
        '''Here we build a new ExecutionPlan from a pipeline definition and the environment config.

        To do this, we iterate through the pipeline's solids in topological order, and hand off the
        execution steps for each solid to a companion _PlanBuilder object.

        Once we've processed the entire pipeline, we invoke _PlanBuilder.build() to construct the
        ExecutionPlan object.
        '''
        check.inst_param(pipeline_def, 'pipeline_def', PipelineDefinition)
        check.inst_param(environment_config, 'environment_config', EnvironmentConfig)

        plan_builder = _PlanBuilder()

        for solid in solids_in_topological_order(pipeline_def):
            ### 1. INPUTS
            # Create and add execution plan steps for solid inputs
            step_inputs = []
            for input_def in solid.definition.input_defs:
                prev_step_output_handle = get_input_source_step_handle(
                    pipeline_def, environment_config, plan_builder, solid, input_def
                )

                # We return None for the handle (see above in get_input_source_step_handle) when the
                # input def runtime type is "Nothing"
                if not prev_step_output_handle:
                    continue

                subplan = create_subplan_for_input(
                    pipeline_def, environment_config, solid, prev_step_output_handle, input_def
                )

                plan_builder.add_steps(subplan.steps)

                step_inputs.append(
                    StepInput(
                        input_def.name, input_def.runtime_type, subplan.terminal_step_output_handle
                    )
                )

            ### 2. TRANSFORM FUNCTION
            # Create and add execution plan step for the solid transform function
            solid_transform_step = create_transform_step(
                pipeline_def, environment_config, solid, step_inputs
            )
            plan_builder.add_step(solid_transform_step)

            ### 3. OUTPUTS
            # Create and add execution plan steps (and output handles) for solid outputs
            for output_def in solid.definition.output_defs:
                subplan = create_subplan_for_output(
                    pipeline_def, environment_config, solid, solid_transform_step, output_def
                )
                plan_builder.add_steps(subplan.steps)

                output_handle = solid.output_handle(output_def.name)
                plan_builder.set_output_handle(output_handle, subplan.terminal_step_output_handle)

        # Finally, we build and return the execution plan
        return plan_builder.build(
            pipeline_def=pipeline_def,
            artifacts_persisted=environment_config.storage.construct_run_storage().is_persistent,
        )
def build(pipeline_def, environment_config):
        '''Here we build a new ExecutionPlan from a pipeline definition and the environment config.

        To do this, we iterate through the pipeline's solids in topological order, and hand off the
        execution steps for each solid to a companion _PlanBuilder object.

        Once we've processed the entire pipeline, we invoke _PlanBuilder.build() to construct the
        ExecutionPlan object.
        '''
        check.inst_param(pipeline_def, 'pipeline_def', PipelineDefinition)
        check.inst_param(environment_config, 'environment_config', EnvironmentConfig)

        plan_builder = _PlanBuilder()

        for solid in solids_in_topological_order(pipeline_def):
            ### 1. INPUTS
            # Create and add execution plan steps for solid inputs
            step_inputs = []
            for input_def in solid.definition.input_defs:
                prev_step_output_handle = get_input_source_step_handle(
                    pipeline_def, environment_config, plan_builder, solid, input_def
                )

                # We return None for the handle (see above in get_input_source_step_handle) when the
                # input def runtime type is "Nothing"
                if not prev_step_output_handle:
                    continue

                subplan = create_subplan_for_input(
                    pipeline_def, environment_config, solid, prev_step_output_handle, input_def
                )

                plan_builder.add_steps(subplan.steps)

                step_inputs.append(
                    StepInput(
                        input_def.name, input_def.runtime_type, subplan.terminal_step_output_handle
                    )
                )

            ### 2. TRANSFORM FUNCTION
            # Create and add execution plan step for the solid transform function
            solid_transform_step = create_transform_step(
                pipeline_def, environment_config, solid, step_inputs
            )
            plan_builder.add_step(solid_transform_step)

            ### 3. OUTPUTS
            # Create and add execution plan steps (and output handles) for solid outputs
            for output_def in solid.definition.output_defs:
                subplan = create_subplan_for_output(
                    pipeline_def, environment_config, solid, solid_transform_step, output_def
                )
                plan_builder.add_steps(subplan.steps)

                output_handle = solid.output_handle(output_def.name)
                plan_builder.set_output_handle(output_handle, subplan.terminal_step_output_handle)

        # Finally, we build and return the execution plan
        return plan_builder.build(
            pipeline_def=pipeline_def,
            artifacts_persisted=environment_config.storage.construct_run_storage().is_persistent,
        )
Here we build a new ExecutionPlan from a pipeline definition and the environment config.

        To do this, we iterate through the pipeline's solids in topological order, and hand off the
        execution steps for each solid to a companion _PlanBuilder object.

        Once we've processed the entire pipeline, we invoke _PlanBuilder.build() to construct the
        ExecutionPlan object.