def get_param_values()

in luigi/task.py [0:0]

• 48 lines of code
• 24 McCabe index (conditional complexity)

    def get_param_values(cls, params, args, kwargs):
        """
        Get the values of the parameters from the args and kwargs.

        :param params: list of (param_name, Parameter).
        :param args: positional arguments
        :param kwargs: keyword arguments.
        :returns: list of `(name, value)` tuples, one for each parameter.
        """
        result = {}

        params_dict = dict(params)

        task_family = cls.get_task_family()

        # In case any exceptions are thrown, create a helpful description of how the Task was invoked
        # TODO: should we detect non-reprable arguments? These will lead to mysterious errors
        exc_desc = '%s[args=%s, kwargs=%s]' % (task_family, args, kwargs)

        # Fill in the positional arguments
        positional_params = [(n, p) for n, p in params if p.positional]
        for i, arg in enumerate(args):
            if i >= len(positional_params):
                raise parameter.UnknownParameterException('%s: takes at most %d parameters (%d given)' % (exc_desc, len(positional_params), len(args)))
            param_name, param_obj = positional_params[i]
            result[param_name] = param_obj.normalize(arg)

        # Then the keyword arguments
        for param_name, arg in kwargs.items():
            if param_name in result:
                raise parameter.DuplicateParameterException('%s: parameter %s was already set as a positional parameter' % (exc_desc, param_name))
            if param_name not in params_dict:
                raise parameter.UnknownParameterException('%s: unknown parameter %s' % (exc_desc, param_name))
            result[param_name] = params_dict[param_name].normalize(arg)

        # Then use the defaults for anything not filled in
        for param_name, param_obj in params:
            if param_name not in result:
                try:
                    has_task_value = param_obj.has_task_value(task_family, param_name)
                except Exception as exc:
                    raise ValueError("%s: Error when parsing the default value of '%s'" % (exc_desc, param_name)) from exc
                if not has_task_value:
                    raise parameter.MissingParameterException("%s: requires the '%s' parameter to be set" % (exc_desc, param_name))
                result[param_name] = param_obj.task_value(task_family, param_name)

        def list_to_tuple(x):
            """ Make tuples out of lists and sets to allow hashing """
            if isinstance(x, list) or isinstance(x, set):
                return tuple(x)
            else:
                return x

        # Check for unconsumed parameters
        conf = configuration.get_config()
        if not hasattr(cls, "_unconsumed_params"):
            cls._unconsumed_params = set()
        if task_family in conf.sections():
            ignore_unconsumed = getattr(cls, 'ignore_unconsumed', set())
            for key, value in conf[task_family].items():
                key = key.replace('-', '_')
                composite_key = f"{task_family}_{key}"
                if key not in result and key not in ignore_unconsumed and composite_key not in cls._unconsumed_params:
                    warnings.warn(
                        "The configuration contains the parameter "
                        f"'{key}' with value '{value}' that is not consumed by the task "
                        f"'{task_family}'.",
                        UnconsumedParameterWarning,
                    )
                    cls._unconsumed_params.add(composite_key)

        # Sort it by the correct order and make a list
        return [(param_name, list_to_tuple(result[param_name])) for param_name, param_obj in params]