def get_help_arguments()

in mephisto/client/cli.py [0:0]

• 103 lines of code
• 23 McCabe index (conditional complexity)

def get_help_arguments(args):
    if len(args) == 0:
        click.echo(
            "Usage: mephisto wut <abstraction>[=<type>] [...specific args to check]"
        )
        return

    from mephisto.operations.registry import (
        get_blueprint_from_type,
        get_crowd_provider_from_type,
        get_architect_from_type,
        get_valid_blueprint_types,
        get_valid_provider_types,
        get_valid_architect_types,
    )
    from mephisto.operations.utils import get_extra_argument_dicts, get_task_state_dicts
    from textwrap import wrap

    VALID_ABSTRACTIONS = ["blueprint", "architect", "requester", "provider", "task"]

    abstraction_equal_split = args[0].split("=", 1)
    abstraction = abstraction_equal_split[0]

    if abstraction not in VALID_ABSTRACTIONS:
        click.echo(
            f"Given abstraction {abstraction} not in valid abstractions {VALID_ABSTRACTIONS}"
        )
        return

    if abstraction == "task":
        from mephisto.data_model.task_config import TaskConfig

        target_class = TaskConfig
    else:
        if len(abstraction_equal_split) == 1:
            # querying about the general abstraction
            if abstraction == "blueprint":
                click.echo(
                    f"The blueprint determines the task content. Valid blueprints are {get_valid_blueprint_types()}"
                )
                return
            elif abstraction == "architect":
                click.echo(
                    f"The architect determines the server where a task is hosted. Valid architects are {get_valid_architect_types()}"
                )
                return
            elif abstraction == "requester":
                click.echo(
                    f"The requester is an account for a crowd provider. Valid requester types are {get_valid_provider_types()}. \n"
                    "Use `mephisto requesters` to see registered requesters, and `mephisto register <requester type>` to register."
                )
                return
            elif abstraction == "provider":
                click.echo(
                    f"The crowd provider determines the source of the crowd workers. Valid provider are {get_valid_provider_types()}"
                )
                return

        # There's a specific abstraction to check
        abstract_value = abstraction_equal_split[1]
        target_class = None
        valid = None
        if abstraction == "blueprint":
            try:
                target_class = get_blueprint_from_type(abstract_value)
            except:
                valid = get_valid_blueprint_types()
        elif abstraction == "architect":
            try:
                target_class = get_architect_from_type(abstract_value)
            except:
                valid = get_valid_architect_types()
        elif abstraction == "provider":
            try:
                target_class = get_crowd_provider_from_type(abstract_value)
            except:
                valid = get_valid_provider_types()
        elif abstraction == "requester":
            try:
                target_class = get_crowd_provider_from_type(
                    abstract_value
                ).RequesterClass
            except:
                valid = get_valid_provider_types()
        if valid is not None:
            click.echo(
                f"The valid types for {abstraction} are {valid}. '{abstract_value}' not found."
            )
            return

    from tabulate import tabulate

    def wrap_fields(in_dict):
        return {
            out_key: {
                in_key: "\n".join(wrap(str(in_val), width=40))
                for in_key, in_val in out_val.items()
            }
            for out_key, out_val in in_dict.items()
        }

    arg_dict = get_extra_argument_dicts(target_class)[0]
    click.echo(arg_dict["desc"])
    checking_args = arg_dict["args"]
    if len(args) > 1:
        checking_args = {k: v for k, v in checking_args.items() if k in args[1:]}
    click.echo(tabulate(wrap_fields(checking_args).values(), headers="keys"))
    if abstraction == "blueprint":
        click.echo(
            f"Additional SharedTaskState args from {target_class.SharedStateClass.__name__}, which may be configured in your run script"
        )
        state_args = get_task_state_dicts(target_class)[0]["args"]
        if len(args) > 1:
            state_args = {k: v for k, v in state_args.items() if k in args[1:]}
        click.echo(tabulate(wrap_fields(state_args).values(), headers="keys"))