PipelineFilterOptions¶

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}