Class DependantScriptValuesProvider

src/config/script/list_values.py:54–102

class DependantScriptValuesProvider(ValuesProvider):

    def __init__(self, script, parameters_supplier, shell, process_invoker: ProcessInvoker) -> None:
        pattern = re.compile('\${([^}]+)\}')

        search_start = 0
        script_template = ''
        required_parameters = set()

        while search_start < len(script):
            match = pattern.search(script, search_start)
            if not match:
                script_template += script[search_start:]
                break
            param_start = match.start()
            if param_start > search_start:
                script_template += script[search_start:param_start]

            param_name = match.group(1)
            required_parameters.add(param_name)

            search_start = match.end() + 1

        self._required_parameters = tuple(required_parameters)
        self._script_template = script
        self._parameters_supplier = parameters_supplier
        self._shell = shell
        self._process_invoker = process_invoker

    def get_required_parameters(self):
        return self._required_parameters

    def get_values(self, parameter_values):
        for param_name in self._required_parameters:
            value_wrapper = parameter_values.get(param_name)
            if (value_wrapper is None) or is_empty(value_wrapper.mapped_script_value):
                return []

        parameters = self._parameters_supplier()
        script = fill_parameter_values(parameters, self._script_template, parameter_values)

        try:
            script_output = self._process_invoker.invoke(script, shell=self._shell)
        except Exception as e:
            LOGGER.warning('Failed to execute script. ' + str(e))
            return []

        script_output = script_output.rstrip('\n')
        return [line for line in script_output.split('\n') if not is_empty(line)]


class FilesProvider(ValuesProvider):