Skip to content

ComputeConfig

kedro_azureml_pipeline.config.ComputeConfig

Bases: RootModel[dict[str, ClusterConfig]]

Named compute clusters with a mandatory __default__ entry.

Use resolve(tag) to look up a cluster by node tag, falling back to __default__.

See Also

ClusterConfig : Single cluster entry. AzureMLPipelineGenerator : Uses compute config for node routing.

Source Code

View on GitHub

Show/Hide source 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
class ComputeConfig(RootModel[dict[str, ClusterConfig]]):
    """Named compute clusters with a mandatory ``__default__`` entry.

    Use ``resolve(tag)`` to look up a cluster by node tag,
    falling back to ``__default__``.

    See Also
    --------
    [ClusterConfig][kedro_azureml_pipeline.config.ClusterConfig] : Single cluster entry.
    [AzureMLPipelineGenerator][kedro_azureml_pipeline.generator.AzureMLPipelineGenerator] : Uses compute config for node routing.
    """

    @model_validator(mode="after")
    def _validate_default_key(self) -> "ComputeConfig":
        """Ensure a ``__default__`` compute entry is present.

        Returns
        -------
        ComputeConfig
            The validated instance.

        Raises
        ------
        ValueError
            If the ``__default__`` key is missing.
        """
        if "__default__" not in self.root:
            raise ValueError("ComputeConfig must contain a '__default__' key")
        return self

    def resolve(self, tag: str | None = None) -> ClusterConfig:
        """Return the cluster for *tag*, falling back to ``__default__``.

        Parameters
        ----------
        tag : str or None
            Node tag to look up. Falls back to ``__default__``
            when ``None``.

        Returns
        -------
        ClusterConfig
            The resolved cluster configuration.
        """
        if tag and tag in self.root:
            return self.root["__default__"].model_copy(update=self.root[tag].model_dump(exclude_none=True))
        return self.root["__default__"]

Methods

resolve(tag=None)

Return the cluster for tag, falling back to __default__.

Parameters
Name Type Description Default
tag str or None

Node tag to look up. Falls back to __default__ when None.

 None
Returns
Type Description
ClusterConfig

The resolved cluster configuration.
Source Code

View on GitHub

Show/Hide source 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def resolve(self, tag: str | None = None) -> ClusterConfig:
    """Return the cluster for *tag*, falling back to ``__default__``.

    Parameters
    ----------
    tag : str or None
        Node tag to look up. Falls back to ``__default__``
        when ``None``.

    Returns
    -------
    ClusterConfig
        The resolved cluster configuration.
    """
    if tag and tag in self.root:
        return self.root["__default__"].model_copy(update=self.root[tag].model_dump(exclude_none=True))
    return self.root["__default__"]