def parse_config()

in share/config.py [0:0]

• 82 lines of code
• 35 McCabe index (conditional complexity)

def parse_config(config_yaml: str, expanders: list[Callable[[str], str]] = []) -> Config:
    """
    Config component factory
    Given a config yaml as string it return the Config instance as defined by the yaml
    """

    for expander in expanders:
        config_yaml = expander(config_yaml)

    yaml_config = yaml.safe_load(config_yaml)
    assert isinstance(yaml_config, dict)

    conf: Config = Config()

    if "inputs" not in yaml_config or not isinstance(yaml_config["inputs"], list):
        raise ValueError("`inputs` must be provided as list")

    for input_n, input_config in enumerate(yaml_config["inputs"]):
        if "id" not in input_config or not isinstance(input_config["id"], str):
            raise ValueError(f"`id` must be provided as string for input at position {input_n + 1}")

        if "type" not in input_config or not isinstance(input_config["type"], str):
            raise ValueError(f'`type` must be provided as string for input {input_config["id"]}')

        try:
            current_input: Input = Input(input_type=input_config["type"], input_id=input_config["id"])
        except ValueError as e:
            raise ValueError(f'An error occurred while applying type configuration for input {input_config["id"]}: {e}')

        if "tags" in input_config:
            current_input.tags = input_config["tags"]

        if "multiline" in input_config:
            if not isinstance(input_config["multiline"], dict):
                raise ValueError(f'`multiline` must be provided as dictionary for input {input_config["id"]}')

            multiline_config = input_config["multiline"]
            if "type" not in multiline_config or not isinstance(multiline_config["type"], str):
                raise ValueError(
                    f'`type` must be provided as string in multiline configuration for input {input_config["id"]}'
                )

            multiline_config["multiline_type"] = multiline_config["type"]
            del multiline_config["type"]

            try:
                current_input.add_multiline_processor(**multiline_config)
            except ValueError as e:
                raise ValueError(
                    f'An error occurred while applying multiline configuration for input {input_config["id"]}: {e}'
                )

        if "expand_event_list_from_field" in input_config:
            current_input.expand_event_list_from_field = input_config["expand_event_list_from_field"]

        if "root_fields_to_add_to_expanded_event" in input_config:
            current_input.root_fields_to_add_to_expanded_event = input_config["root_fields_to_add_to_expanded_event"]

        if "json_content_type" in input_config:
            current_input.json_content_type = input_config["json_content_type"]

        include_rules: list[IncludeExcludeRule] = []
        if "include" in input_config:
            include_rules_from_config = input_config["include"]
            if not isinstance(include_rules_from_config, list):
                raise ValueError(f'`include` must be provided as list for input {input_config["id"]}')

            for include_rule in include_rules_from_config:
                include_rules.append(IncludeExcludeRule(pattern=str(include_rule)))

        exclude_rules: list[IncludeExcludeRule] = []
        if "exclude" in input_config:
            exclude_rules_from_config = input_config["exclude"]
            if not isinstance(exclude_rules_from_config, list):
                raise ValueError(f'`exclude` must be provided as list for input {input_config["id"]}')

            for exclude_rule in exclude_rules_from_config:
                exclude_rules.append(IncludeExcludeRule(pattern=str(exclude_rule)))

        if len(include_rules) > 0 or len(exclude_rules) > 0:
            current_input.include_exclude_filter = IncludeExcludeFilter(
                include_patterns=include_rules, exclude_patterns=exclude_rules
            )

        if "outputs" not in input_config or not isinstance(input_config["outputs"], list):
            raise ValueError(f'`outputs` must be provided as list for input {input_config["id"]}')

        for output_n, output_config in enumerate(input_config["outputs"]):
            if "type" not in output_config or not isinstance(output_config["type"], str):
                raise ValueError(
                    f"`type` for output configuration at position {output_n + 1} must "
                    f'be provided as string for input {input_config["id"]}'
                )

            if "args" not in output_config or not isinstance(output_config["args"], dict):
                raise ValueError(
                    f"`args` for output configuration at position {output_n + 1} "
                    f'must be provided as dictionary for input {input_config["id"]}'
                )

            output_config["args"]["tags"] = current_input.tags

            try:
                current_input.add_output(output_type=output_config["type"], **output_config["args"])
            except ValueError as e:
                raise ValueError(
                    f"An error occurred while applying output configuration "
                    f'at position {output_n + 1} for input {input_config["id"]}: {e}'
                )

        conf.add_input(current_input)

    return conf