def config_options_with_config_input()

in metaflow/user_configs/config_options.py [0:0]

• 88 lines of code
• 15 McCabe index (conditional complexity)

def config_options_with_config_input(cmd):
    help_strs = []
    required_names = []
    defaults = {}
    config_seen = set()
    parsers = {}
    flow_cls = getattr(current_flow, "flow_cls", None)
    if flow_cls is None:
        return cmd, None

    parameters = [p for _, p in flow_cls._get_parameters() if p.IS_CONFIG_PARAMETER]
    # List all the configuration options
    for arg in parameters[::-1]:
        kwargs = arg.option_kwargs(False)
        if arg.name.lower() in config_seen:
            msg = (
                "Multiple configurations use the same name '%s'. Note that names are "
                "case-insensitive. Please change the "
                "names of some of your configurations" % arg.name
            )
            raise MetaflowException(msg)
        config_seen.add(arg.name.lower())
        if kwargs["required"]:
            required_names.append(arg.name)

        defaults[arg.name.lower()] = (
            arg.kwargs.get("default", None),
            arg._default_is_file,
        )
        help_strs.append("  - %s: %s" % (arg.name.lower(), kwargs.get("help", "")))
        parsers[arg.name.lower()] = arg.parser

    if not config_seen:
        # No configurations -- don't add anything; we set it to False so that it
        # can be checked whether or not we called this.
        return cmd, False

    help_str = (
        "Configuration options for the flow. "
        "Multiple configurations can be specified. Cannot be used with resume."
    )
    help_str = "\n\n".join([help_str] + help_strs)
    config_input = ConfigInput(required_names, defaults, parsers)
    cb_func = config_input.process_configs_click

    cmd.params.insert(
        0,
        click.Option(
            ["--config-value", "config_value"],
            nargs=2,
            multiple=True,
            type=MultipleTuple([click.Choice(config_seen), ConvertDictOrStr()]),
            callback=cb_func,
            help=help_str,
            envvar="METAFLOW_FLOW_CONFIG_VALUE",
            show_default=False,
            default=[
                (
                    k,
                    (
                        ConvertDictOrStr.mark_as_default(v[0])
                        if not callable(v[0]) and not v[1]
                        else None
                    ),
                )
                for k, v in defaults.items()
            ],
            required=False,
        ),
    )
    cmd.params.insert(
        0,
        click.Option(
            ["--config", "config"],
            nargs=2,
            multiple=True,
            type=MultipleTuple([click.Choice(config_seen), ConvertPath()]),
            callback=cb_func,
            help=help_str,
            envvar="METAFLOW_FLOW_CONFIG",
            show_default=False,
            default=[
                (
                    k,
                    (
                        ConvertPath.mark_as_default(v[0])
                        if not callable(v[0]) and v[1]
                        else None
                    ),
                )
                for k, v in defaults.items()
            ],
            required=False,
        ),
    )
    return cmd, config_input