python_modules/dagster/dagster/core/definitions/pipeline.py
def _build_sub_pipeline(pipeline_def, solid_names):
    '''
    Build a pipeline which is a subset of another pipeline.
    Only includes the solids which are in solid_names.
    '''

    check.inst_param(pipeline_def, 'pipeline_def', PipelineDefinition)
    check.list_param(solid_names, 'solid_names', of_type=str)

    solid_name_set = set(solid_names)
    solids = list(map(pipeline_def.solid_named, solid_names))
    deps = {_dep_key_of(solid): {} for solid in solids}

    def _out_handle_of_inp(input_handle):
        if pipeline_def.dependency_structure.has_dep(input_handle):
            output_handle = pipeline_def.dependency_structure.get_dep(input_handle)
            if output_handle.solid.name in solid_name_set:
                return output_handle
        return None

    for solid in solids:
        for input_handle in solid.input_handles():
            output_handle = _out_handle_of_inp(input_handle)
            if output_handle:
                deps[_dep_key_of(solid)][input_handle.input_def.name] = DependencyDefinition(
                    solid=output_handle.solid.name, output=output_handle.output_def.name
                )

    return PipelineDefinition(
        name=pipeline_def.name,
        solids=list({solid.definition for solid in solids}),
        context_definitions=pipeline_def.context_definitions,
        dependencies=deps,
    )
def _build_sub_pipeline(pipeline_def, solid_names):
    '''
    Build a pipeline which is a subset of another pipeline.
    Only includes the solids which are in solid_names.
    '''

    check.inst_param(pipeline_def, 'pipeline_def', PipelineDefinition)
    check.list_param(solid_names, 'solid_names', of_type=str)

    solid_name_set = set(solid_names)
    solids = list(map(pipeline_def.solid_named, solid_names))
    deps = {_dep_key_of(solid): {} for solid in solids}

    def _out_handle_of_inp(input_handle):
        if pipeline_def.dependency_structure.has_dep(input_handle):
            output_handle = pipeline_def.dependency_structure.get_dep(input_handle)
            if output_handle.solid.name in solid_name_set:
                return output_handle
        return None

    for solid in solids:
        for input_handle in solid.input_handles():
            output_handle = _out_handle_of_inp(input_handle)
            if output_handle:
                deps[_dep_key_of(solid)][input_handle.input_def.name] = DependencyDefinition(
                    solid=output_handle.solid.name, output=output_handle.output_def.name
                )

    return PipelineDefinition(
        name=pipeline_def.name,
        solids=list({solid.definition for solid in solids}),
        context_definitions=pipeline_def.context_definitions,
        dependencies=deps,
    )
def _build_sub_pipeline ( pipeline_def , solid_names ) : check . inst_param ( pipeline_def , 'pipeline_def' , PipelineDefinition ) check . list_param ( solid_names , 'solid_names' , of_type = str ) solid_name_set = set ( solid_names ) solids = list ( map ( pipeline_def . solid_named , solid_names ) ) deps = { _dep_key_of ( solid ) : { } for solid in solids } def _out_handle_of_inp ( input_handle ) : if pipeline_def . dependency_structure . has_dep ( input_handle ) : output_handle = pipeline_def . dependency_structure . get_dep ( input_handle ) if output_handle . solid . name in solid_name_set : return output_handle return None for solid in solids : for input_handle in solid . input_handles ( ) : output_handle = _out_handle_of_inp ( input_handle ) if output_handle : deps [ _dep_key_of ( solid ) ] [ input_handle . input_def . name ] = DependencyDefinition ( solid = output_handle . solid . name , output = output_handle . output_def . name ) return PipelineDefinition ( name = pipeline_def . name , solids = list ( { solid . definition for solid in solids } ) , context_definitions = pipeline_def . context_definitions , dependencies = deps , )
Build a pipeline which is a subset of another pipeline.
    Only includes the solids which are in solid_names.
Build a pipeline which is a subset of another pipeline . Only includes the solids which are in solid_names .