Skip to content

Runtime parameters

RuntimeParameter = Annotated[Union[_T, None], Field(default=None), _RUNTIME_PARAMETER_ANNOTATION] module-attribute

Used to mark the attributes of a Step as a runtime parameter.

RuntimeParametersMixin

Bases: BaseModel

Mixin for classes that have RuntimeParameters attributes.

Attributes:

Name Type Description
_runtime_parameters Dict[str, Any]

A dictionary containing the values of the runtime parameters of the class. This attribute is meant to be used internally and should not be accessed directly.
Source code in src/distilabel/mixins/runtime_parameters.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
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
class RuntimeParametersMixin(BaseModel):
    """Mixin for classes that have `RuntimeParameter`s attributes.

    Attributes:
        _runtime_parameters: A dictionary containing the values of the runtime parameters
            of the class. This attribute is meant to be used internally and should not be
            accessed directly.
    """

    _runtime_parameters: Dict[str, Any] = PrivateAttr(default_factory=dict)

    @property
    def runtime_parameters_names(self) -> RuntimeParametersNames:
        """Returns a dictionary containing the name of the runtime parameters of the class
        as keys and whether the parameter is required or not as values.

        Returns:
            A dictionary containing the name of the runtime parameters of the class as keys
            and whether the parameter is required or not as values.
        """

        runtime_parameters = {}

        for name, field_info in self.model_fields.items():  # type: ignore
            is_runtime_param, is_optional = _is_runtime_parameter(field_info)
            if is_runtime_param:
                runtime_parameters[name] = is_optional
                continue

            attr = getattr(self, name)
            if isinstance(attr, RuntimeParametersMixin):
                runtime_parameters[name] = attr.runtime_parameters_names

        return runtime_parameters

    def get_runtime_parameters_info(self) -> List[Dict[str, Any]]:
        """Gets the information of the runtime parameters of the class such as the name and
        the description. This function is meant to include the information of the runtime
        parameters in the serialized data of the class.

        Returns:
            A list containing the information for each runtime parameter of the class.
        """
        runtime_parameters_info = []
        for name, field_info in self.model_fields.items():  # type: ignore
            if name not in self.runtime_parameters_names:
                continue

            attr = getattr(self, name)
            if isinstance(attr, RuntimeParametersMixin):
                runtime_parameters_info.append(
                    {
                        "name": name,
                        "runtime_parameters_info": attr.get_runtime_parameters_info(),
                    }
                )
                continue

            info = {"name": name, "optional": self.runtime_parameters_names[name]}
            if field_info.description is not None:
                info["description"] = field_info.description
            runtime_parameters_info.append(info)
        return runtime_parameters_info

    def set_runtime_parameters(self, runtime_parameters: Dict[str, Any]) -> None:
        """Sets the runtime parameters of the class using the provided values. If the attr
        to be set is a `RuntimeParametersMixin`, it will call `set_runtime_parameters` on
        the attr.

        Args:
            runtime_parameters: A dictionary containing the values of the runtime parameters
                to set.
        """
        runtime_parameters_names = list(self.runtime_parameters_names.keys())
        for name, value in runtime_parameters.items():
            if name not in self.runtime_parameters_names:
                # Check done just to ensure the unit tests for the mixin run
                if getattr(self, "pipeline", None):
                    closest = difflib.get_close_matches(
                        name, runtime_parameters_names, cutoff=0.5
                    )
                    msg = (
                        f"⚠️  Runtime parameter '{name}' unknown in step '{self.name}'."
                    )
                    if closest:
                        msg += f" Did you mean any of: {closest}"
                    else:
                        msg += f" Available runtime parameters for the step: {runtime_parameters_names}."
                    self.pipeline._logger.warning(msg)
                continue

            attr = getattr(self, name)
            if isinstance(attr, RuntimeParametersMixin):
                attr.set_runtime_parameters(value)
                self._runtime_parameters[name] = value
                continue

            # Handle settings values for `_SecretField`
            field_info = self.model_fields[name]
            inner_type = _extract_runtime_parameter_inner_type(field_info.annotation)
            if inspect.isclass(inner_type) and issubclass(inner_type, _SecretField):
                value = inner_type(value)

            setattr(self, name, value)
            self._runtime_parameters[name] = value

runtime_parameters_names: RuntimeParametersNames property

Returns a dictionary containing the name of the runtime parameters of the class as keys and whether the parameter is required or not as values.

Returns:

Type Description
RuntimeParametersNames

A dictionary containing the name of the runtime parameters of the class as keys
RuntimeParametersNames

and whether the parameter is required or not as values.

get_runtime_parameters_info()

Gets the information of the runtime parameters of the class such as the name and the description. This function is meant to include the information of the runtime parameters in the serialized data of the class.

Returns:

Type Description
List[Dict[str, Any]]

A list containing the information for each runtime parameter of the class.
Source code in src/distilabel/mixins/runtime_parameters.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def get_runtime_parameters_info(self) -> List[Dict[str, Any]]:
    """Gets the information of the runtime parameters of the class such as the name and
    the description. This function is meant to include the information of the runtime
    parameters in the serialized data of the class.

    Returns:
        A list containing the information for each runtime parameter of the class.
    """
    runtime_parameters_info = []
    for name, field_info in self.model_fields.items():  # type: ignore
        if name not in self.runtime_parameters_names:
            continue

        attr = getattr(self, name)
        if isinstance(attr, RuntimeParametersMixin):
            runtime_parameters_info.append(
                {
                    "name": name,
                    "runtime_parameters_info": attr.get_runtime_parameters_info(),
                }
            )
            continue

        info = {"name": name, "optional": self.runtime_parameters_names[name]}
        if field_info.description is not None:
            info["description"] = field_info.description
        runtime_parameters_info.append(info)
    return runtime_parameters_info

set_runtime_parameters(runtime_parameters)

Sets the runtime parameters of the class using the provided values. If the attr to be set is a RuntimeParametersMixin, it will call set_runtime_parameters on the attr.

Parameters:

Name Type Description Default
runtime_parameters Dict[str, Any]

A dictionary containing the values of the runtime parameters to set.

 required
Source code in src/distilabel/mixins/runtime_parameters.py 
100
101
102
103
104
105
106
107
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
def set_runtime_parameters(self, runtime_parameters: Dict[str, Any]) -> None:
    """Sets the runtime parameters of the class using the provided values. If the attr
    to be set is a `RuntimeParametersMixin`, it will call `set_runtime_parameters` on
    the attr.

    Args:
        runtime_parameters: A dictionary containing the values of the runtime parameters
            to set.
    """
    runtime_parameters_names = list(self.runtime_parameters_names.keys())
    for name, value in runtime_parameters.items():
        if name not in self.runtime_parameters_names:
            # Check done just to ensure the unit tests for the mixin run
            if getattr(self, "pipeline", None):
                closest = difflib.get_close_matches(
                    name, runtime_parameters_names, cutoff=0.5
                )
                msg = (
                    f"⚠️  Runtime parameter '{name}' unknown in step '{self.name}'."
                )
                if closest:
                    msg += f" Did you mean any of: {closest}"
                else:
                    msg += f" Available runtime parameters for the step: {runtime_parameters_names}."
                self.pipeline._logger.warning(msg)
            continue

        attr = getattr(self, name)
        if isinstance(attr, RuntimeParametersMixin):
            attr.set_runtime_parameters(value)
            self._runtime_parameters[name] = value
            continue

        # Handle settings values for `_SecretField`
        field_info = self.model_fields[name]
        inner_type = _extract_runtime_parameter_inner_type(field_info.annotation)
        if inspect.isclass(inner_type) and issubclass(inner_type, _SecretField):
            value = inner_type(value)

        setattr(self, name, value)
        self._runtime_parameters[name] = value