kedro/pipeline/modular_pipeline.py
def pipeline(
    pipe: Pipeline,
    *,
    inputs: Dict[str, str] = None,
    outputs: Dict[str, str] = None,
    parameters: Dict[str, str] = None,
    namespace: str = None,
) -> Pipeline:
    """Create a copy of the pipeline and its nodes,
    with some dataset names and node names modified.

    Args:
        pipe: Original modular pipeline to integrate
        inputs: A map of the existing input name to the new one.
            Must only refer to the pipeline's free inputs.
        outputs: A map of the existing output name to the new one.
            Can refer to both the pipeline's free outputs, as well
            as intermediate results that need to be exposed.
        parameters: A map of existing parameter to the new one.
        namespace: A prefix to give to all dataset names,
            except those explicitly named with the `inputs`/`outputs`
            arguments, and parameter references (`params:` and `parameters`).

    Raises:
        ModularPipelineError: When inputs, outputs or parameters are incorrectly
            specified, or they do not exist on the original pipeline.
        ValueError: When underlying pipeline nodes inputs/outputs are not
            any of the expected types (str, dict, list, or None).

    Returns:
        A new ``Pipeline`` object with the new nodes, modified as requested.
    """
    # pylint: disable=protected-access
    inputs = copy.deepcopy(inputs) or {}
    outputs = copy.deepcopy(outputs) or {}
    parameters = copy.deepcopy(parameters) or {}

    _validate_datasets_exist(inputs.keys(), outputs.keys(), parameters.keys(), pipe)
    _validate_inputs_outputs(inputs.keys(), outputs.keys(), pipe)

    mapping = {**inputs, **outputs, **parameters}

    def _prefix(name: str) -> str:
        return "{}.{}".format(namespace, name) if namespace else name

    def _is_transcode_base_in_mapping(name: str) -> bool:
        base_name, _ = _transcode_split(name)
        return base_name in mapping

    def _map_transcode_base(name: str):
        base_name, transcode_suffix = _transcode_split(name)
        return TRANSCODING_SEPARATOR.join((mapping[base_name], transcode_suffix))

    def _rename(name: str):
        rules = [
            # if name mapped to new name, update with new name
            (lambda n: n in mapping, lambda n: mapping[n]),
            # if it's a parameter, leave as is (don't namespace)
            (_is_parameter, lambda n: n),
            # if transcode base is mapped to a new name, update with new base
            (_is_transcode_base_in_mapping, _map_transcode_base),
            # if namespace given, prefix name using that namespace
            (lambda n: bool(namespace), _prefix),
        ]

        for predicate, processor in rules:
            if predicate(name):
                return processor(name)

        # leave name as is
        return name

    def _process_dataset_names(
        datasets: Union[None, str, List[str], Dict[str, str]]
    ) -> Union[None, str, List[str], Dict[str, str]]:
        if datasets is None:
            return None
        if isinstance(datasets, str):
            return _rename(datasets)
        if isinstance(datasets, list):
            return [_rename(name) for name in datasets]
        if isinstance(datasets, dict):
            return {key: _rename(value) for key, value in datasets.items()}

        raise ValueError(  # pragma: no cover
            "Unexpected input {} of type {}".format(datasets, type(datasets))
        )

    def _copy_node(node: Node) -> Node:
        new_namespace = node.namespace
        if namespace:
            new_namespace = (
                f"{namespace}.{node.namespace}" if node.namespace else namespace
            )

        return node._copy(
            inputs=_process_dataset_names(node._inputs),
            outputs=_process_dataset_names(node._outputs),
            namespace=new_namespace,
        )

    new_nodes = [_copy_node(n) for n in pipe.nodes]

    return Pipeline(new_nodes)
Create a copy of the pipeline and its nodes,
    with some dataset names and node names modified.

    Args:
        pipe: Original modular pipeline to integrate
        inputs: A map of the existing input name to the new one.
            Must only refer to the pipeline's free inputs.
        outputs: A map of the existing output name to the new one.
            Can refer to both the pipeline's free outputs, as well
            as intermediate results that need to be exposed.
        parameters: A map of existing parameter to the new one.
        namespace: A prefix to give to all dataset names,
            except those explicitly named with the `inputs`/`outputs`
            arguments, and parameter references (`params:` and `parameters`).

    Raises:
        ModularPipelineError: When inputs, outputs or parameters are incorrectly
            specified, or they do not exist on the original pipeline.
        ValueError: When underlying pipeline nodes inputs/outputs are not
            any of the expected types (str, dict, list, or None).

    Returns:
        A new ``Pipeline`` object with the new nodes, modified as requested.
    
Create a copy of the pipeline and its nodes,
with some dataset names and node names modified.
Create a copy of the pipeline and its nodes with some dataset names and node names modified .
def pipeline(
    pipe: Pipeline,
    *,
    inputs: Dict[str, str] = None,
    outputs: Dict[str, str] = None,
    parameters: Dict[str, str] = None,
    namespace: str = None,
) -> Pipeline:
    inputs = copy.deepcopy(inputs) or {}
    outputs = copy.deepcopy(outputs) or {}
    parameters = copy.deepcopy(parameters) or {}
    _validate_datasets_exist(inputs.keys(), outputs.keys(), parameters.keys(), pipe)
    _validate_inputs_outputs(inputs.keys(), outputs.keys(), pipe)
    mapping = {**inputs, **outputs, **parameters}
    def _prefix(name: str) -> str:
        return "{}.{}".format(namespace, name) if namespace else name
    def _is_transcode_base_in_mapping(name: str) -> bool:
        base_name, _ = _transcode_split(name)
        return base_name in mapping
    def _map_transcode_base(name: str):
        base_name, transcode_suffix = _transcode_split(name)
        return TRANSCODING_SEPARATOR.join((mapping[base_name], transcode_suffix))
    def _rename(name: str):
        rules = [
            (lambda n: n in mapping, lambda n: mapping[n]),
            (_is_parameter, lambda n: n),
            (_is_transcode_base_in_mapping, _map_transcode_base),
            (lambda n: bool(namespace), _prefix),
        ]
        for predicate, processor in rules:
            if predicate(name):
                return processor(name)
        return name
    def _process_dataset_names(
        datasets: Union[None, str, List[str], Dict[str, str]]
    ) -> Union[None, str, List[str], Dict[str, str]]:
        if datasets is None:
            return None
        if isinstance(datasets, str):
            return _rename(datasets)
        if isinstance(datasets, list):
            return [_rename(name) for name in datasets]
        if isinstance(datasets, dict):
            return {key: _rename(value) for key, value in datasets.items()}
        raise ValueError(  
            "Unexpected input {} of type {}".format(datasets, type(datasets))
        )
    def _copy_node(node: Node) -> Node:
        new_namespace = node.namespace
        if namespace:
            new_namespace = (
                f"{namespace}.{node.namespace}" if node.namespace else namespace
            )
        return node._copy(
            inputs=_process_dataset_names(node._inputs),
            outputs=_process_dataset_names(node._outputs),
            namespace=new_namespace,
        )
    new_nodes = [_copy_node(n) for n in pipe.nodes]
    return Pipeline(new_nodes)
def pipeline ( pipe : Pipeline , * , inputs : Dict [ str , str ] = None , outputs : Dict [ str , str ] = None , parameters : Dict [ str , str ] = None , namespace : str = None , ) -> Pipeline : inputs = copy . deepcopy ( inputs ) or { } outputs = copy . deepcopy ( outputs ) or { } parameters = copy . deepcopy ( parameters ) or { } _validate_datasets_exist ( inputs . keys ( ) , outputs . keys ( ) , parameters . keys ( ) , pipe ) _validate_inputs_outputs ( inputs . keys ( ) , outputs . keys ( ) , pipe ) mapping = { ** inputs , ** outputs , ** parameters } def _prefix ( name : str ) -> str : return "{}.{}" . format ( namespace , name ) if namespace else name def _is_transcode_base_in_mapping ( name : str ) -> bool : base_name , _ = _transcode_split ( name ) return base_name in mapping def _map_transcode_base ( name : str ) : base_name , transcode_suffix = _transcode_split ( name ) return TRANSCODING_SEPARATOR . join ( ( mapping [ base_name ] , transcode_suffix ) ) def _rename ( name : str ) : rules = [ ( lambda n : n in mapping , lambda n : mapping [ n ] ) , ( _is_parameter , lambda n : n ) , ( _is_transcode_base_in_mapping , _map_transcode_base ) , ( lambda n : bool ( namespace ) , _prefix ) , ] for predicate , processor in rules : if predicate ( name ) : return processor ( name ) return name def _process_dataset_names ( datasets : Union [ None , str , List [ str ] , Dict [ str , str ] ] ) -> Union [ None , str , List [ str ] , Dict [ str , str ] ] : if datasets is None : return None if isinstance ( datasets , str ) : return _rename ( datasets ) if isinstance ( datasets , list ) : return [ _rename ( name ) for name in datasets ] if isinstance ( datasets , dict ) : return { key : _rename ( value ) for key , value in datasets . items ( ) } raise ValueError ( "Unexpected input {} of type {}" . format ( datasets , type ( datasets ) ) ) def _copy_node ( node : Node ) -> Node : new_namespace = node . namespace if namespace : new_namespace = ( f"{namespace}.{node.namespace}" if node . namespace else namespace ) return node . _copy ( inputs = _process_dataset_names ( node . _inputs ) , outputs = _process_dataset_names ( node . _outputs ) , namespace = new_namespace , ) new_nodes = [ _copy_node ( n ) for n in pipe . nodes ] return Pipeline ( new_nodes )
Create a copy of the pipeline and its nodes,
with some dataset names and node names modified.
Create a copy of the pipeline and its nodes with some dataset names and node names modified .
"""Create a copy of the pipeline and its nodes,
    with some dataset names and node names modified.

    Args:
        pipe: Original modular pipeline to integrate
        inputs: A map of the existing input name to the new one.
            Must only refer to the pipeline's free inputs.
        outputs: A map of the existing output name to the new one.
            Can refer to both the pipeline's free outputs, as well
            as intermediate results that need to be exposed.
        parameters: A map of existing parameter to the new one.
        namespace: A prefix to give to all dataset names,
            except those explicitly named with the `inputs`/`outputs`
            arguments, and parameter references (`params:` and `parameters`).

    Raises:
        ModularPipelineError: When inputs, outputs or parameters are incorrectly
            specified, or they do not exist on the original pipeline.
        ValueError: When underlying pipeline nodes inputs/outputs are not
            any of the expected types (str, dict, list, or None).

    Returns:
        A new ``Pipeline`` object with the new nodes, modified as requested.
    """ # pylint: disable=protected-access # if name mapped to new name, update with new name # if it's a parameter, leave as is (don't namespace) # if transcode base is mapped to a new name, update with new base # if namespace given, prefix name using that namespace # leave name as is # pragma: no cover
A new ``Pipeline`` object with the new nodes, modified as requested.
A new ` ` Pipeline ` ` object with the new nodes modified as requested .
When inputs, outputs or parameters are incorrectly
specified, or they do not exist on the original pipeline.
When inputs outputs or parameters are incorrectly specified or they do not exist on the original pipeline .
ModularPipelineError
When underlying pipeline nodes inputs/outputs are not
any of the expected types (str, dict, list, or None).
When underlying pipeline nodes inputs / outputs are not any of the expected types ( str dict list or None ) .
ValueError
pipe
Pipeline
Original modular pipeline to integrate
Original modular pipeline to integrate
inputs
Dict[str, str]
A map of the existing input name to the new one.
Must only refer to the pipeline's free inputs.
A map of the existing input name to the new one . Must only refer to the pipeline ' s free inputs .
outputs
Dict[str, str]
A map of the existing output name to the new one.
Can refer to both the pipeline's free outputs, as well
as intermediate results that need to be exposed.
A map of the existing output name to the new one . Can refer to both the pipeline ' s free outputs as well as intermediate results that need to be exposed .
parameters
Dict[str, str]
A map of existing parameter to the new one.
A map of existing parameter to the new one .
namespace
str
A prefix to give to all dataset names,
except those explicitly named with the `inputs`/`outputs`
arguments, and parameter references (`params:` and `parameters`).
A prefix to give to all dataset names except those explicitly named with the ` inputs ` / ` outputs ` arguments and parameter references ( ` params : ` and ` parameters ` ) .