cli

CliCommands dataclass

Container to store the commands and subcommands for the cli.

Source code in m/cli/engine/types.py

@dataclass
class CliCommands:
    """Container to store the commands and subcommands for the cli."""

    commands: dict[str, CommandModule | CliSubcommands]

    # Optional root meta data to provide information about the cli.
    meta: MetaModule | None

CommandModule dataclass

Container to store the run function from a "command" module.

Source code in m/cli/engine/types.py

@dataclass
class CommandModule:
    """Container to store the run function from a "command" module."""

    run: DecoratedRunFunction

FuncArgs dataclass

Stores function arguments.

Source code in m/cli/engine/types.py

@dataclass
class FuncArgs:
    """Stores function arguments."""

    args: list[Any]
    kwargs: dict[str, Any]

MetaModule dataclass

Container to store a metadata dictionary from a "meta" module.

Source code in m/cli/engine/types.py

@dataclass
class MetaModule:
    """Container to store a metadata dictionary from a "meta" module."""

    meta: dict[str, str]
    add_arguments: Callable[[ap.ArgumentParser], None] | None = None

Arg(default=PydanticUndefined, *, help, positional=None, required=None, aliases=None, nargs=None, validator=None)

Create a pydantic Field.

Field docs: https://docs.pydantic.dev/2.2/usage/fields

Defines properties used to create an argparse argument. This function should work for most cases. If we need something that is not covered we can use ArgProxy instead which is untyped but provides all the arguments and keyword arguments to argparse.

See https://docs.python.org/3/library/argparse.html.

Parameters:

Name	Type	Description	Default
default	Any	Default value if the field is not set.	PydanticUndefined
help	str	Human-readable description.	required
positional	bool | None	Whether the argument is positional or not.	None
required	bool | None	Indicate whether an argument is required or optional.	None
aliases	list[str] | None	Alternative names for the argument.	None
nargs	int | Literal['?', '*', '+'] | None	Number of times the argument can be used.	None
validator	Callable[[str], str] | None	Function to validate the argument.	None

Returns:

Type	Description
Any	A new FieldInfo. The return annotation is Any so Arg can be used on type annotated fields without causing a typing error.

Source code in m/cli/args.py

def Arg(  # noqa: N802, WPS211
    default: Any = PydanticUndefined,
    *,
    help: str,  # noqa: WPS125
    positional: bool | None = None,
    required: bool | None = None,
    aliases: list[str] | None = None,
    nargs: int | Literal['?', '*', '+'] | None = None,
    validator: Callable[[str], str] | None = None,
) -> Any:
    """Create a pydantic `Field`.

    Field docs: https://docs.pydantic.dev/2.2/usage/fields

    Defines properties used to create an argparse argument. This function
    should work for most cases. If we need something that is not covered
    we can use `ArgProxy` instead which is untyped but provides all the arguments
    and keyword arguments to argparse.

    See https://docs.python.org/3/library/argparse.html.

    Args:
        default: Default value if the field is not set.
        help: Human-readable description.
        positional: Whether the argument is positional or not.
        required: Indicate whether an argument is required or optional.
        aliases: Alternative names for the argument.
        nargs: Number of times the argument can be used.
        validator: Function to validate the argument.

    Returns:
        A new [`FieldInfo`][pydantic.fields.FieldInfo]. The return annotation is
            `Any` so `Arg` can be used on type annotated fields without causing
            a typing error.
    """
    extras = {
        'positional': positional,
        'validator': validator,
        'aliases': aliases,
        'nargs': nargs,
        'required': required,
    }
    return FieldInfo.from_field(
        default,
        description=cleandoc(help),
        json_schema_extra={k: v for k, v in extras.items() if v is not None},
    )

ArgProxy(*args, **kwargs)

Wrap function to provide all argparse inputs.

This is a escape hatch and does not provide any typings. See https://docs.python.org/3/library/argparse.html.

Parameters:

Name	Type	Description	Default
args	Any	The arguments to argparse add_arguments.	()
kwargs	Any	The keyword arguments to argparse add arguments.	{}

Returns:

Type	Description
Any	A new FieldInfo. The return annotation is Any so Arg can be used on type annotated fields without causing a typing error.

Source code in m/cli/args.py

def ArgProxy(*args: Any, **kwargs: Any) -> Any:  # noqa: N802
    """Wrap function to provide all argparse inputs.

    This is a escape hatch and does not provide any typings.
    See https://docs.python.org/3/library/argparse.html.

    Args:
        args: The arguments to argparse add_arguments.
        kwargs: The keyword arguments to argparse add arguments.

    Returns:
        A new [`FieldInfo`][pydantic.fields.FieldInfo]. The return annotation is
            `Any` so `Arg` can be used on type annotated fields without causing
            a typing error.
    """
    return FieldInfo.from_field(
        json_schema_extra={
            'proxy': FuncArgs(args=list(args), kwargs=kwargs),
        },
    )

Meta(*, help, description)

Create the meta dictionary for a subcommand description.

In the case of the root command the help may be set to empty since it is not used.

Parameters:

Name	Type	Description	Default
help	str	The help message for the subcommand.	required
description	str	The description for the command/subcommand.	required

Returns:

Type	Description
dict[str, str]	A dictionary with the help and description.

Source code in m/cli/args.py

def Meta(  # noqa: N802
    *,
    help: str,  # noqa: WPS125
    description: str,
) -> dict[str, str]:
    """Create the meta dictionary for a subcommand description.

    In the case of the root command the `help` may be set to empty since it
    is not used.

    Args:
        help: The help message for the subcommand.
        description: The description for the command/subcommand.

    Returns:
        A dictionary with the help and description.
    """
    return {
        'help': help,
        'description': cleandoc(description),
    }

RemainderArgs(*, help=_Unset)

Provide a list of unrecognized arguments.

This is a escape hatch and does not provide any typings. May be useful for commands that need to pass arguments to other commands.

Parameters:

Name	Type	Description	Default
help	str | None	Human-readable description.	_Unset

Returns:

Type	Description
Any	A new FieldInfo, the return annotation is
Any	Any so Arg can be used on type annotated fields without causing a typing error.

Source code in m/cli/args.py

def RemainderArgs(  # noqa: N802
    *,
    help: str | None = _Unset,  # noqa: WPS125
) -> Any:
    """Provide a list of unrecognized arguments.

    This is a escape hatch and does not provide any typings. May be
    useful for commands that need to pass arguments to other commands.

    Args:
        help: Human-readable description.

    Returns:
        A new [`FieldInfo`][pydantic.fields.FieldInfo], the return annotation is
        `Any` so `Arg` can be used on type annotated fields without causing a typing error.
    """
    return FieldInfo.from_field(
        [],
        description=help,
        json_schema_extra={'__remainder_args': True},
    )

add_arg(*args, **kwargs)

Wrap FuncArgs arguments in a function.

Parameters:

Name	Type	Description	Default
args	Any	The arguments to argparse add_arguments.	()
kwargs	Any	The keyword arguments to argparse add arguments.	{}

Returns:

Type	Description
FuncArgs	A FuncArgs instance.

Source code in m/cli/engine/types.py

@typing_extensions.deprecated(
    'The `add_arg` method is deprecated; use `m.cli.ArgProxy` instead.',
)
def add_arg(*args: Any, **kwargs: Any) -> FuncArgs:
    """Wrap FuncArgs arguments in a function.

    Args:
        args: The arguments to argparse add_arguments.
        kwargs: The keyword arguments to argparse add arguments.

    Returns:
        A FuncArgs instance.
    """
    # `m` does not reference this function anymore, excluding from coverage
    return FuncArgs(args=list(args), kwargs=kwargs)  # pragma: no cover

cli_commands(root_meta=None, **commands)

Create a CliCommands instance.

Parameters:

Name	Type	Description	Default
root_meta	MetaModule | None	The meta for the root command.	None
commands	DecoratedRunFunction | CliSubcommands	The commands and subcommands for the cli.	{}

Returns:

Type	Description
CliCommands	An instance of CliCommands.

Source code in m/cli/engine/sys.py

def cli_commands(
    root_meta: MetaModule | None = None,
    **commands: DecoratedRunFunction | CliSubcommands,
) -> CliCommands:
    """Create a CliCommands instance.

    Args:
        root_meta: The meta for the root command.
        commands: The commands and subcommands for the cli.

    Returns:
        An instance of CliCommands.
    """
    root_meta = root_meta or default_root_meta
    return CliCommands(
        meta=root_meta,
        commands={
            cmd_name: (
                cmd_item
                if isinstance(cmd_item, CliSubcommands)
                else CommandModule(run=cmd_item)
            )
            for cmd_name, cmd_item in commands.items()
        },
    )

cli_integration_token(integration, env_var)

Return a function that takes in a parser.

This generated function registers a token argument in the parser which looks for its value in the environment variables.

Parameters:

Name	Type	Description	Default
integration	str	The name of the integration.	required
env_var	str	The environment variable name.	required

Returns:

Type	Description
cli_global_option	A function to add arguments to an argparse parser.

Source code in m/cli/cli.py

def cli_integration_token(
    integration: str,
    env_var: str,
) -> cli_global_option:
    """Return a function that takes in a parser.

    This generated function registers a token argument in the parser
    which looks for its value in the environment variables.

    Args:
        integration: The name of the integration.
        env_var: The environment variable name.

    Returns:
        A function to add arguments to an argparse parser.
    """
    return lambda parser: parser.add_argument(
        '-t',
        '--token',
        type=validate_non_empty_str,
        default=env(env_var),
        help=f'{integration} access token (default: env.{env_var})',
    )

command(*, help, model, name=None)

Apply a decorator to the run function to make it into a command.

Parameters:

Name	Type	Description	Default
name	str | None	The command name.	None
help	str	A short description of the command.	required
model	type[BaseModel]	A pydantic model to describe the cli arguments.	required

Returns:

Type	Description
partial[partial[int]]	A transformed run function aware of the arguments model.

Source code in m/cli/engine/argparse.py

def command(
    *,
    help: str,  # noqa: WPS125
    model: type[BaseModel],
    name: str | None = None,
) -> partial[partial[int]]:
    """Apply a decorator to the `run` function to make it into a command.

    Args:
        name: The command name.
        help: A short description of the command.
        model: A pydantic model to describe the cli arguments.

    Returns:
        A transformed run function aware of the arguments model.
    """
    # m no longer uses the name argument but we keep it for now
    if name:  # pragma: no cover
        warn('`name` is no longer needed, please remove it', DeprecationWarning)
    return partial(_handle_decorated_func, CommandInputs(help, model))

command_group(*, help, description, add_arguments=None)

Create an instance of a MetaModule.

Named cmd_group since it is used to describe the group of commands.

Parameters:

Name	Type	Description	Default
help	str	quick help describing the module.	required
description	str	Detailed description about the module.	required
add_arguments	Callable[[ArgumentParser], None] | None	Optional function to handle the argparse instance.	None

Returns:

Type	Description
MetaModule	An instance of a MetaModule.

Source code in m/cli/engine/sys.py

def command_group(
    *,
    help: str,   # noqa: WPS125
    description: str,
    add_arguments: Callable[[ap.ArgumentParser], None] | None = None,
) -> MetaModule:
    """Create an instance of a MetaModule.

    Named `cmd_group` since it is used to describe the group of commands.

    Args:
        help: quick help describing the module.
        description: Detailed description about the module.
        add_arguments: Optional function to handle the argparse instance.

    Returns:
        An instance of a MetaModule.
    """
    return MetaModule(
        meta=Meta(help=help, description=description),
        add_arguments=add_arguments,
    )

create_issue_handler(use_warning)

Generate a function to log an issue.

Parameters:

Name	Type	Description	Default
use_warning	bool	Uses a warning log instead of an error.	required

Returns:

Type	Description
Callable[[Issue], None]	A function that uses the arguments provided.

Source code in m/cli/handlers.py

def create_issue_handler(use_warning: bool) -> Callable[[Issue], None]:
    """Generate a function to log an issue.

    Args:
        use_warning: Uses a warning log instead of an error.

    Returns:
        A function that uses the arguments provided.
    """
    return partial(_issue_handler, use_warning)

create_json_handler(pretty)

Generate a function to display a value to stdout as json.

Parameters:

Name	Type	Description	Default
pretty	bool	If true, it formats with indentation of 2 spaces.	required

Returns:

Type	Description
Callable[[Any], None]	A function that uses the arguments provided.

Source code in m/cli/handlers.py

def create_json_handler(pretty: bool) -> Callable[[Any], None]:
    """Generate a function to display a value to stdout as json.

    Args:
        pretty: If true, it formats with indentation of 2 spaces.

    Returns:
        A function that uses the arguments provided.
    """
    return partial(_json_handler, pretty)

create_yaml_handler(pretty)

Generate a function to display a value to stdout as yaml.

Parameters:

Name	Type	Description	Default
pretty	bool	If true, it highlights the output.	required

Returns:

Type	Description
Callable[[Any], None]	A function that uses the arguments provided.

Source code in m/cli/handlers.py

def create_yaml_handler(pretty: bool) -> Callable[[Any], None]:
    """Generate a function to display a value to stdout as yaml.

    Args:
        pretty: If true, it highlights the output.

    Returns:
        A function that uses the arguments provided.
    """
    return partial(_yaml_handler, pretty)

env_var(arg_value)

Read a value from the environment.

Parameters:

Name	Type	Description	Default
arg_value	str	The input provided by argparse.	required

Returns:

Type	Description
str	The environment variable value if it exists. Otherwise, the input.

Source code in m/cli/validators.py

def env_var(arg_value: str) -> str:
    """Read a value from the environment.

    Args:
        arg_value: The input provided by argparse.

    Returns:
        The environment variable value if it exists. Otherwise, the input.
    """
    return os.environ.get(arg_value, arg_value)

env_var_or_empty(arg_value)

Read a value from the environment.

Unlike the env_var validator, this will only allow the arg_value to pass through if is not in the form of an environment variable. That is, if the value is all uppercase letters and underscores it will attempt to read from the environment and return an empty string if not defined.

Parameters:

Name	Type	Description	Default
arg_value	str	The input provided by argparse.	required

Returns:

Type	Description
str	The environment variable value if it exists. Otherwise, empty string.

Source code in m/cli/validators.py

def env_var_or_empty(arg_value: str) -> str:
    """Read a value from the environment.

    Unlike the [env_var][m.cli.validators.env_var] validator, this will only
    allow the `arg_value` to pass through if is not in the form of an
    environment variable. That is, if the value is all uppercase letters and
    underscores it will attempt to read from the environment and return an empty
    string if not defined.

    Args:
        arg_value: The input provided by argparse.

    Returns:
        The environment variable value if it exists. Otherwise, empty string.
    """
    env_form = arg_value.upper().replace('-', '_', -1)
    if arg_value == env_form:
        return os.environ.get(arg_value, '')
    return arg_value

exec_cli(cli_commands)

Execute the cli application.

usage::

def create_cli_commands() -> CliCommands:
    # We may import CliCommand objects from other projects and create
    # a new one with them.
    return import_cli_commands('cli.command.module')

def main():
    cli_commands = create_cli_commands()
    exec_cli(cli_commands)

This is the preferred way to execute the cli application as it will allow other potential applications to use the cli commands.

Parameters:

Name	Type	Description	Default
cli_commands	CliCommands	The cli commands to execute.	required

Source code in m/cli/cli.py

def exec_cli(cli_commands: CliCommands) -> None:
    """Execute the cli application.

    usage::

        def create_cli_commands() -> CliCommands:
            # We may import CliCommand objects from other projects and create
            # a new one with them.
            return import_cli_commands('cli.command.module')

        def main():
            cli_commands = create_cli_commands()
            exec_cli(cli_commands)

    This is the preferred way to execute the cli application as it will allow
    other potential applications to use the cli commands.

    Args:
        cli_commands: The cli commands to execute.
    """
    arg = _main_parser(cli_commands)

    run_func = None
    command_name = ''

    commands = cli_commands.commands
    # WPS421 encourages to use try/except instead of hasattr but in this
    # case we want explicitly before using it.
    if hasattr(arg, 'subcommand_name'):  # noqa: WPS421
        command_name = arg.subcommand_name
        sub_mod = cast(CliSubcommands, commands[arg.command_name])
        run_func = sub_mod.subcommands[command_name].run
    else:
        command_name = arg.command_name
        run_func = cast(CommandModule, commands[command_name]).run

    len_run_args = params_count(run_func)
    run_args = [command_name, arg, None][:len_run_args]
    exit_code = 0
    try:
        # mypy is having a hard time figuring out the type of run_args
        exit_code = run_func(*run_args)  # type:ignore[arg-type]
    except Exception as ex:
        default_issue_handler(
            Issue('unknown cli run function exception', cause=ex),
        )
        exit_code = 5
    sys.exit(exit_code)

import_cli_commands(commands_module)

Gather the commands and subcommands for the cli.

Parameters:

Name	Type	Description	Default
commands_module	str	module containing all the commands.	required

Returns:

Type	Description
CliCommands	An instance of CliCommands gathered from the commands_module.

Source code in m/cli/engine/sys.py

def import_cli_commands(commands_module: str) -> CliCommands:
    """Gather the commands and subcommands for the cli.

    Args:
        commands_module: module containing all the commands.

    Returns:
        An instance of CliCommands gathered from the commands_module.
    """
    cmd_module = import_module(commands_module)
    if hasattr(cmd_module, 'create_cli_commands'):
        # No extra checks for now, if a module has this we assume that
        # it has the right signature.
        return cast(CliCommands, cmd_module.create_cli_commands())

    commands: dict[str, CommandModule | CliSubcommands] = {}
    root_cmd = get_command_modules(commands_module)

    for key, cmd_mod in root_cmd.items():
        commands[key] = cmd_mod

    root = pth.split(cmd_module.__file__ or '')[0]
    for cmd_name in iglob(f'{root}/*'):
        if cmd_name.endswith('.py') or cmd_name.endswith('__'):
            continue
        name = pth.split(cmd_name)[1]
        commands[name] = CliSubcommands(
            meta=_get_meta_module(f'{commands_module}.{name}', default_meta),
            subcommands=get_command_modules(f'{commands_module}.{name}'),
        )

    return CliCommands(
        meta=_get_meta_module(commands_module, default_root_meta),
        commands=commands,
    )

merge_cli_commands(base, overrides, **resolutions)

Merge two CliCommands instances.

Resolutions may be provided to resolve merge conflicts between two subcommands.

Parameters:

Name	Type	Description	Default
base	CliCommands	The base CliCommands instance.	required
overrides	CliCommands	The overrides CliCommands instance.	required
resolutions	SubCmdResolution	A dictionary of resolutions for subcommands.	{}

Returns:

Type	Description
CliCommands	A new CliCommands instance.

Source code in m/cli/engine/sys.py

def merge_cli_commands(
    base: CliCommands,
    overrides: CliCommands,
    **resolutions: SubCmdResolution,
) -> CliCommands:
    """Merge two CliCommands instances.

    Resolutions may be provided to resolve merge conflicts between two
    subcommands.

    Args:
        base: The base CliCommands instance.
        overrides: The overrides CliCommands instance.
        resolutions: A dictionary of resolutions for subcommands.

    Returns:
        A new CliCommands instance.
    """
    commands: dict[str, CommandModule | CliSubcommands] = {**base.commands}
    for cmd_name, cmd_item in overrides.commands.items():
        resolution = resolutions.get(cmd_name)
        commands[cmd_name] = _get_new_command(cmd_name, commands, cmd_item, resolution)

    return CliCommands(
        meta=base.meta or overrides.meta,
        commands=commands,
    )

run_cli(commands_module, add_args=None)

Run the cli application.

Deprecated, use exec_cli instead.

usage::

def add_args(argp):
    argp.add_argument(...)
def main():
    run_cli('m.cli.commands', add_args)

We only need add_args if we need to gain access to the argparse.ArgumentParser instance.

Parameters:

Name	Type	Description	Default
commands_module	str | None	The full name of the module containing the commands.	required
add_args	Callable[[ArgumentParser], None] | None	Optional callback to gain access to the ArgumentParser.	None

Source code in m/cli/cli.py

@typing_extensions.deprecated(
    '`run_cli` deprecated; use `exec_cli` instead.',
)
def run_cli(
    commands_module: str | None,
    add_args: Callable[[argparse.ArgumentParser], None] | None = None,
) -> None:  # pragma: no cover
    """Run the cli application.

    Deprecated, use `exec_cli` instead.

    usage::

        def add_args(argp):
            argp.add_argument(...)
        def main():
            run_cli('m.cli.commands', add_args)

    We only need `add_args` if we need to gain access to the
    `argparse.ArgumentParser` instance.

    Args:
        commands_module: The full name of the module containing the commands.
        add_args: Optional callback to gain access to the ArgumentParser.
    """
    # NOTE: This is a deprecated feature and will be removed in the future.
    if commands_module and '/' in commands_module:
        warn(
            '`run_cli(__file__)` is deprecated, use `run_cli("commands.module") instead',
            DeprecationWarning,
        )
        root = pth.split(pth.split(commands_module)[0])[1]
        commands_module = f'{root}.cli.commands'
    cli_commands: CliCommands = CliCommands(commands={}, meta=default_root_meta)
    if commands_module:
        cli_commands = import_cli_commands(commands_module)
    if add_args:
        warn('run_cli add_args is deprecated, use meta.add_arguments instead', DeprecationWarning)
        if not cli_commands.meta:
            cli_commands.meta = MetaModule(
                meta=default_root_meta.meta,
                add_arguments=add_args,
            )
        cli_commands.meta.add_arguments = add_args
    exec_cli(cli_commands)

run_main(callback, result_handler=default_result_handler, issue_handler=default_issue_handler)

Run the callback and print the returned value.

To change how the result or an issue should be display provide the optional arguments handle_result and handle_issue. For instance, to display the raw value provide the print function.

Parameters:

Name	Type	Description	Default
callback	Callable[[], OneOf[Issue, Any]]	A function that returns a OneOf.	required
result_handler	Callable[[Any], None]	A function that takes in the Good result.	default_result_handler
issue_handler	Callable[[Issue], None]	A function that takes in the Issue.	default_issue_handler

Returns:

Type	Description
int	0 if the callback is a Good result otherwise return 1.

Source code in m/cli/cli.py

def run_main(
    callback: Callable[[], OneOf[Issue, Any]],
    result_handler: Callable[[Any], None] = default_result_handler,
    issue_handler: Callable[[Issue], None] = default_issue_handler,
) -> int:
    """Run the callback and print the returned value.

    To change how the result or an issue should be display provide the optional
    arguments `handle_result` and `handle_issue`. For instance, to display the
    raw value provide the `print` function.

    Args:
        callback: A function that returns a `OneOf`.
        result_handler: A function that takes in the Good result.
        issue_handler: A function that takes in the Issue.

    Returns:
        0 if the callback is a `Good` result otherwise return 1.
    """
    res = None
    try:
        res = callback()
    except Exception as ex:
        issue_handler(Issue('unknown caught exception', cause=ex))
        return 2
    if isinstance(res, Bad):
        problem = res.value
        if isinstance(problem, Issue):
            issue_handler(problem)
        else:
            issue_handler(Issue('non-issue exception', cause=problem))
        return 1
    result_handler(res.value)
    return 0

subcommands(meta=None, **commands)

Create a CliSubcommands instance.

Parameters:

Name	Type	Description	Default
meta	MetaModule | None	The meta for the command group.	None
commands	DecoratedRunFunction	The commands the group.	{}

Returns:

Type	Description
CliSubcommands	An instance of CliSubcommands.

Source code in m/cli/engine/sys.py

def subcommands(
    meta: MetaModule | None = None,
    **commands: DecoratedRunFunction,
) -> CliSubcommands:
    """Create a CliSubcommands instance.

    Args:
        meta: The meta for the command group.
        commands: The commands the group.

    Returns:
        An instance of CliSubcommands.
    """
    return CliSubcommands(
        meta=meta,
        subcommands={
            cmd_name: CommandModule(run=cmd_item)
            for cmd_name, cmd_item in commands.items()
        },
    )

validate_file_exists(path)

Assert that a file exists.

Parameters:

Name	Type	Description	Default
path	str	The path to the file.	required

Raises:

Type	Description
ArgumentTypeError	If the file does not exist.

Returns:

Type	Description
str	The path if it exists.

Source code in m/cli/validators.py

def validate_file_exists(path: str) -> str:
    """Assert that a file exists.

    Args:
        path: The path to the file.

    Raises:
        ArgumentTypeError: If the file does not exist.

    Returns:
        The path if it exists.
    """
    path_inst = Path(path)
    if path_inst.exists():
        return path
    raise ArgumentTypeError('file does not exist')

validate_json_payload(file_path)

Return a dictionary from the contents of file_path.

This is a string that tell us to read from a file, stdin or just plain json data.

It can be used parse yaml files as well. The extension should be .yaml or .yml.

Parameters:

Name	Type	Description	Default
file_path	str	A string with the json payload. If it starts with @ then the name of a valid json file from where the payload will be read.	required

Raises:

Type	Description
ArgumentTypeError	If the file_path is meant to be a valid path and it does not exist.

Returns:

Type	Description
Any	A parsed json payload

Source code in m/cli/validators.py

def validate_json_payload(file_path: str) -> Any:
    """Return a dictionary from the contents of file_path.

    This is a string that tell us to read from a file, stdin or just
    plain json data.

    It can be used parse `yaml` files as well. The extension should be
    `.yaml` or `.yml`.

    Args:
        file_path: A string with the json payload. If it starts with `@` then
            the name of a valid json file from where the payload will be read.

    Raises:
        ArgumentTypeError: If the file_path is meant to be a valid path and it
            does not exist.

    Returns:
        A parsed json payload
    """
    if file_path == '@-':
        res = read_json(None)
        if res.is_bad:
            msg = f'invalid json payload in SYS.STDIN\n{res.value}'
            raise ArgumentTypeError(msg)
        return res.value
    if file_path.startswith('@'):
        err = ''
        filename = file_path[1:]
        if Path.exists(Path(filename)):
            res = read_yson(filename)
            if not res.is_bad:
                return res.value
            err = f'invalid json payload in {filename}\n{res.value}'
        else:
            err = f'file "{filename}" does not exist'
        if err:
            raise ArgumentTypeError(err)
    res = parse_json(file_path)
    if res.is_bad:
        raise ArgumentTypeError(f'invalid json payload\n{res.value}')
    return res.value

validate_payload(file_path)

Return the raw payload.

This allows us to read from a file or the stdin stream.

Parameters:

Name	Type	Description	Default
file_path	str	A string with the payload. If it starts with @ then the name of a valid file from where the payload will be read.	required

Raises:

Type	Description
ArgumentTypeError	If the file_path is meant to be a valid path and it does not exist.

Returns:

Type	Description
str	The payload found in the file.

Source code in m/cli/validators.py

def validate_payload(file_path: str) -> str:
    """Return the raw payload.

    This allows us to read from a file or the stdin stream.

    Args:
        file_path: A string with the payload. If it starts with `@` then the
            name of a valid file from where the payload will be read.

    Raises:
        ArgumentTypeError: If the file_path is meant to be a valid path and it
            does not exist.

    Returns:
        The payload found in the file.
    """
    if file_path.startswith(r'\@'):
        # escape @ with \ to let the cli know that the payload starts with @
        return file_path[1:]
    if file_path == '@-':
        return sys.stdin.read()
    if file_path.startswith('@'):
        filename = file_path[1:]
        path_handle = Path(filename)
        if not Path.exists(path_handle):
            raise ArgumentTypeError(f'file "{filename}" does not exist')
        with Path.open(path_handle, encoding='UTF-8') as fp:
            return fp.read()
    return file_path