confit.cli

Cli

Bases: Typer

Custom Typer object that:

• validates a command parameters before executing it
• accepts a configuration file describing the parameters
• automatically instantiates parameters given a dictionary when type hinted

Source code in confit/cli.py

class Cli(Typer):
    """
    Custom Typer object that:

    - validates a command parameters before executing it
    - accepts a configuration file describing the parameters
    - automatically instantiates parameters given a dictionary when type hinted
    """

    def command(  # noqa
        self,
        name,
        *,
        cls: Optional[Type[TyperCommand]] = None,
        context_settings: Optional[Dict[Any, Any]] = None,
        help: Optional[str] = None,
        epilog: Optional[str] = None,
        short_help: Optional[str] = None,
        options_metavar: str = "[OPTIONS]",
        add_help_option: bool = True,
        no_args_is_help: bool = False,
        hidden: bool = False,
        deprecated: bool = False,
        # Rich settings
        rich_help_panel: Union[str, None] = Default(None),
        registry: Any = None,
        default_config: Optional[Config] = None,
        merge_with_default_config: bool = False,
    ) -> Callable[[CommandFunctionType], CommandFunctionType]:
        typer_command = super().command(
            name=name,
            cls=cls,
            help=help,
            epilog=epilog,
            short_help=short_help,
            options_metavar=options_metavar,
            add_help_option=add_help_option,
            no_args_is_help=no_args_is_help,
            hidden=hidden,
            deprecated=deprecated,
            rich_help_panel=rich_help_panel,
            context_settings={
                **(context_settings or {}),
                "ignore_unknown_options": True,
                "allow_extra_args": True,
            },
        )

        def wrapper(fn):
            validated = validate_arguments(fn)

            @typer_command
            def command(ctx: Context, config: Optional[List[Path]] = None):
                config_path = config

                has_meta = _fn_has_meta(fn)
                if config_path:
                    config, name_from_file = merge_from_disk(config_path)
                elif default_config is not None:
                    config = default_config
                else:
                    config = Config({name: {}})
                if default_config is not None and merge_with_default_config:
                    config = Config(default_config).merge(config)
                model_fields = (
                    validated.model.model_fields
                    if hasattr(validated.model, "model_fields")
                    else validated.model.__fields__
                )
                for k, v in parse_overrides(ctx.args).items():
                    if "." not in k:
                        parts = (name, k)
                    else:
                        parts = k.split(".")
                        if parts[0] in model_fields and parts[0] not in config:
                            parts = (name, *parts)
                    current = config
                    if parts[0] not in current:
                        raise Exception(
                            f"{k} does not match any existing section in config"
                        )
                    for part in parts[:-1]:
                        current = current.setdefault(part, Config())
                    current[parts[-1]] = v
                try:
                    default_seed = model_fields.get("seed")
                    if default_seed is not None:
                        default_seed = default_seed.get_default()
                    seed = Config.resolve(
                        config.get(name, {}).get("seed", default_seed),
                        registry=registry,
                        root=config,
                    )
                    if seed is not None:
                        set_seed(seed)
                    resolved_config = Config(config[name]).resolve(
                        registry=registry, root=config
                    )
                    if has_meta:
                        config_meta = dict(
                            config_path=config_path,
                            resolved_config=resolved_config,
                            unresolved_config=config,
                        )
                        return validated(
                            **resolved_config,
                            config_meta=config_meta,
                        )
                    else:
                        return validated(**resolved_config)
                except (LegacyValidationError, ConfitValidationError) as e:
                    e.raw_errors = patch_errors(
                        e.raw_errors,
                        (name,),
                    )
                    if is_debug() or e.__cause__ is not None:
                        raise e

                    def cprint(*args, style=None, **kw):  # pragma: no cover
                        return print(*args, **kw)

                    try:  # pragma: no cover
                        try:
                            from typer.main import console_stderr
                        except ImportError:
                            import rich

                            console_stderr = rich.console.Console(stderr=True)
                        cprint = console_stderr.print
                    except Exception:  # pragma: no cover
                        pass
                    cprint("Validation error:", style="red", end=" ")
                    cprint(str(e))
                    sys.exit(1)
                except KeyboardInterrupt as e:  # pragma: no cover
                    raise Exception("Interrupted by user") from e

            return validated

        return wrapper

parse_overrides(args)

Parse the overrides from the command line into a dictionary of key-value pairs.

PARAMETER	DESCRIPTION
args	The arguments to parse TYPE: List[str]

RETURNS	DESCRIPTION
Dict[str, Any]	The parsed overrides as a dictionary

Source code in confit/cli.py

def parse_overrides(args: List[str]) -> Dict[str, Any]:
    """
    Parse the overrides from the command line into a dictionary
    of key-value pairs.

    Parameters
    ----------
    args: List[str]
        The arguments to parse

    Returns
    -------
    Dict[str, Any]
        The parsed overrides as a dictionary
    """
    result = {}
    while args:
        opt = args.pop(0)
        err = f"Invalid config override '{opt}'"
        if opt.startswith("--"):  # new argument
            opt = opt.replace("--", "")
            if "=" in opt:  # we have --opt=value
                opt, value = opt.split("=", 1)
            else:
                if not args or args[0].startswith("--"):  # flag with no value
                    value = "true"
                else:
                    value = args.pop(0)
            opt = opt.replace("-", "_")
            result[opt] = loads(value)
        else:
            secho(f"{err}: doesn't support shorthands", fg=colors.RED)
            exit(1)
    return result