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,
        )
def build ( pipeline_def , environment_config ) : 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.
Here we build a new ExecutionPlan from a pipeline definition and the environment config .