PipelineFilterOptions

kedro_azureml_pipeline.config.PipelineFilterOptions

Bases: BaseModel

Kedro pipeline filter options for selecting nodes.

Parameters

Name	Type	Description	Default
pipeline_name	str	Kedro pipeline name (default "__default__").	required
from_nodes	list of str or None	Start from these nodes.	required
to_nodes	list of str or None	Run up to these nodes.	required
node_names	list of str or None	Run only these specific nodes.	required
from_inputs	list of str or None	Start from nodes that produce these datasets.	required
to_outputs	list of str or None	Run up to nodes that produce these datasets.	required
node_namespaces	list of str or None	Filter by namespace.	required
tags	list of str or None	Filter by tag.	required

See Also

JobConfig : Uses filter options per job. AzureMLPipelineGenerator : Applies filters during generation.

Source Code

View on GitHub

Show/Hide source

class PipelineFilterOptions(BaseModel):
    """Kedro pipeline filter options for selecting nodes.

    Parameters
    ----------
    pipeline_name : str
        Kedro pipeline name (default ``"__default__"``).
    from_nodes : list of str or None
        Start from these nodes.
    to_nodes : list of str or None
        Run up to these nodes.
    node_names : list of str or None
        Run only these specific nodes.
    from_inputs : list of str or None
        Start from nodes that produce these datasets.
    to_outputs : list of str or None
        Run up to nodes that produce these datasets.
    node_namespaces : list of str or None
        Filter by namespace.
    tags : list of str or None
        Filter by tag.

    See Also
    --------
    [JobConfig][kedro_azureml_pipeline.config.JobConfig] : Uses filter options per job.
    [AzureMLPipelineGenerator][kedro_azureml_pipeline.generator.AzureMLPipelineGenerator] : Applies filters during generation.
    """

    model_config = ConfigDict(extra="forbid")

    pipeline_name: str = Field(default="__default__", description="Kedro pipeline name.")
    from_nodes: list[str] | None = Field(default=None, description="Start from these nodes.")
    to_nodes: list[str] | None = Field(default=None, description="Run up to these nodes.")
    node_names: list[str] | None = Field(default=None, description="Run only these specific nodes.")
    from_inputs: list[str] | None = Field(default=None, description="Start from nodes that produce these datasets.")
    to_outputs: list[str] | None = Field(default=None, description="Run up to nodes that produce these datasets.")
    node_namespaces: list[str] | None = Field(default=None, description="Filter by namespace.")
    tags: list[str] | None = Field(default=None, description="Filter by tag.")

    def to_filter_kwargs(self) -> dict[str, Any]:
        """Return non-None filter kwargs suitable for ``Pipeline.filter()``.

        Returns
        -------
        dict of str to Any
            Only keys whose values are not ``None``.
        """
        mapping = {
            "node_names": self.node_names,
            "from_nodes": self.from_nodes,
            "to_nodes": self.to_nodes,
            "from_inputs": self.from_inputs,
            "to_outputs": self.to_outputs,
            "node_namespaces": self.node_namespaces,
            "tags": self.tags,
        }
        return {k: v for k, v in mapping.items() if v is not None}

Methods

to_filter_kwargs()

Return non-None filter kwargs suitable for Pipeline.filter().

Returns

Type	Description
dict of str to Any	Only keys whose values are not None.

Source Code

View on GitHub

Show/Hide source

def to_filter_kwargs(self) -> dict[str, Any]:
    """Return non-None filter kwargs suitable for ``Pipeline.filter()``.

    Returns
    -------
    dict of str to Any
        Only keys whose values are not ``None``.
    """
    mapping = {
        "node_names": self.node_names,
        "from_nodes": self.from_nodes,
        "to_nodes": self.to_nodes,
        "from_inputs": self.from_inputs,
        "to_outputs": self.to_outputs,
        "node_namespaces": self.node_namespaces,
        "tags": self.tags,
    }
    return {k: v for k, v in mapping.items() if v is not None}