Skip to content

cli_tools

convert_choice(value: str, choices: Collection[str]) -> list[str]

Convert a choice to a list of choices, handling the special 'All' choice.

Parameters

value The choice to convert. choices The set of choices to choose from.

Returns

list[str] The list of choices.

Source code in src/rra_tools/cli_tools/options.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def convert_choice(value: str, choices: Collection[str]) -> list[str]:
    """Convert a choice to a list of choices, handling the special 'All' choice.

    Parameters
    ----------
    value
        The choice to convert.
    choices
        The set of choices to choose from.

    Returns
    -------
    list[str]
        The list of choices.
    """
    if value == RUN_ALL:
        return list(choices)
    elif value in choices:
        return [value]
    else:
        msg = f"Invalid choice: {value}. Must be one of {choices} or {RUN_ALL}."
        raise ValueError(msg)

handle_exceptions(func: Callable[P, T], logger: SupportsLogging, *, with_debugger: bool) -> Callable[P, T]

Drops a user into an interactive debugger if func raises an error.

Source code in src/rra_tools/cli_tools/exceptions.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def handle_exceptions[**P, T](
    func: Callable[P, T],
    logger: SupportsLogging,
    *,
    with_debugger: bool,
) -> Callable[P, T]:
    """Drops a user into an interactive debugger if func raises an error."""

    @functools.wraps(func)
    def wrapped(*args: P.args, **kwargs: P.kwargs) -> T:  # type: ignore[return]
        try:
            return func(*args, **kwargs)
        except (BdbQuit, KeyboardInterrupt):
            raise
        except Exception:
            msg = "Uncaught exception"
            logger.exception(msg)
            if with_debugger:
                import pdb  # noqa: T100
                import traceback

                traceback.print_exc()
                pdb.post_mortem()
            else:
                raise

    return wrapped

import_module_from_info(module_info: ModuleInfo) -> ModuleType

Import a module from a ModuleInfo object.

Source code in src/rra_tools/cli_tools/importers.py 
 5
 6
 7
 8
 9
10
def import_module_from_info(module_info: ModuleInfo) -> ModuleType:
    """Import a module from a ModuleInfo object."""
    finder = module_info.module_finder
    spec = finder.find_spec(module_info.name)  # type: ignore[call-arg]
    module = spec.loader.load_module(module_info.name)  # type: ignore[union-attr]
    return module  # noqa: RET504

process_choices(allow_all: bool, choices: Collection[str] | None) -> tuple[click.ParamType, str | None, bool]

Support function for creating options with choices.

A common pattern in RRA pipelines is to build CLIs that admit a choice of a specific set of values or a special value that represents all possible values. This function provides a way to handle this pattern in a consistent way.

There are four possible cases: 1. No choices are provided and RUN_ALL is allowed. This is useful when the set of choices is not known ahead of time, or is contingent on another option. For example, if there is a task that depends on location and year, but the years available depend on the location. The user might want to run a single year for a location (which they'll have to know ahead of time); or all years for a location, which would be the subset of years available for that location; or all years for all locations, which could be a different subset of years for each included location. 2. Choices are provided and RUN_ALL is allowed. This is useful when the set of choices is known ahead of time, but the user might want to run all of them. 3. No choices are provided and RUN_ALL is not allowed. This is useful when the set of choices is not known ahead of time, but the user must provide a value. 4. Choices are provided and RUN_ALL is not allowed. This is useful when the set of choices is known ahead of time and the user must provide a value.

Parameters

allow_all Whether to allow the special value RUN_ALL. choices The set of choices to allow.

Returns

tuple[click.ParamType, str | None, bool] The option type, default value, and whether to show the default.

Source code in src/rra_tools/cli_tools/options.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def process_choices(
    allow_all: bool,  # noqa: FBT001
    choices: Collection[str] | None,
) -> tuple[click.ParamType, str | None, bool]:
    """Support function for creating options with choices.

    A common pattern in RRA pipelines is to build CLIs that admit a choice
    of a specific set of values or a special value that represents all
    possible values. This function provides a way to handle this pattern
    in a consistent way.

    There are four possible cases:
    1. No choices are provided and RUN_ALL is allowed. This is useful when the
        set of choices is not known ahead of time, or is contingent on another
        option. For example, if there is a task that depends on location and year,
        but the years available depend on the location. The user might want to
        run a single year for a location (which they'll have to know ahead of time);
        or all years for a location, which would be the subset of years available
        for that location; or all years for all locations, which could be a different
        subset of years for each included location.
    2. Choices are provided and RUN_ALL is allowed. This is useful when the set of
        choices is known ahead of time, but the user might want to run all of them.
    3. No choices are provided and RUN_ALL is not allowed. This is useful when the
        set of choices is not known ahead of time, but the user must provide a value.
    4. Choices are provided and RUN_ALL is not allowed. This is useful when the set of
        choices is known ahead of time and the user must provide a value.

    Parameters
    ----------
    allow_all
        Whether to allow the special value RUN_ALL.
    choices
        The set of choices to allow.

    Returns
    -------
    tuple[click.ParamType, str | None, bool]
        The option type, default value, and whether to show the default.
    """

    if choices is None:
        option_type: click.ParamType = click.STRING
        default = RUN_ALL if allow_all else None
    else:
        choices = list(choices)
        if allow_all:
            choices.append(RUN_ALL)
            default = RUN_ALL
        else:
            default = None
        option_type = click.Choice(choices)
    show_default = default is not None
    return option_type, default, show_default

with_choice(name: str, short_name: str | None = None, *, allow_all: bool = True, choices: Collection[str] | None = None, convert: bool | None = None, **kwargs: Any) -> Callable[[Callable[P, T]], Callable[P, T]]

Create an option with a set of choices.

Parameters

name The name of the option. short_name An optional short name for the option. allow_all Whether to allow the special value "ALL", which represents all choices. choices The set of choices to allow. convert Whether to convert the provided argument to a list, resolving the special value "ALL" to all choices.

Source code in src/rra_tools/cli_tools/options.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def with_choice[**P, T](
    name: str,
    short_name: str | None = None,
    *,
    allow_all: bool = True,
    choices: Collection[str] | None = None,
    convert: bool | None = None,
    **kwargs: Any,
) -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Create an option with a set of choices.

    Parameters
    ----------
    name
        The name of the option.
    short_name
        An optional short name for the option.
    allow_all
        Whether to allow the special value "ALL", which represents all choices.
    choices
        The set of choices to allow.
    convert
        Whether to convert the provided argument to a list, resolving the special
        value "ALL" to all choices.

    """

    names = [f"--{name.replace('_', '-')}"]
    if short_name is not None:
        if len(short_name) != 1:
            msg = "Short names must be a single character."
            raise ValueError(msg)
        names.append(f"-{short_name}")
    option_type, default, show_default = process_choices(allow_all, choices)

    if choices and convert is None:
        convert = allow_all

    if convert:
        if not allow_all:
            msg = "Conversion is only supported when allow_all is True."
            raise ValueError(msg)
        if choices is None:
            msg = "Conversion is only supported when choices are provided."
            raise ValueError(msg)

        if "callback" in kwargs:
            old_callback = kwargs.pop("callback")

            def _callback(
                ctx: click.Context,
                param: click.Parameter,
                value: str,
            ) -> list[str]:
                value = old_callback(ctx, param, value)
                return convert_choice(value, choices)
        else:

            def _callback(
                ctx: click.Context,  # noqa: ARG001
                param: click.Parameter,  # noqa: ARG001
                value: str,
            ) -> list[str]:
                return convert_choice(value, choices)

        kwargs["callback"] = _callback

    return click.option(
        *names,
        type=option_type,
        default=default,
        show_default=show_default,
        **kwargs,
    )

exceptions

handle_exceptions(func: Callable[P, T], logger: SupportsLogging, *, with_debugger: bool) -> Callable[P, T]

Drops a user into an interactive debugger if func raises an error.

Source code in src/rra_tools/cli_tools/exceptions.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def handle_exceptions[**P, T](
    func: Callable[P, T],
    logger: SupportsLogging,
    *,
    with_debugger: bool,
) -> Callable[P, T]:
    """Drops a user into an interactive debugger if func raises an error."""

    @functools.wraps(func)
    def wrapped(*args: P.args, **kwargs: P.kwargs) -> T:  # type: ignore[return]
        try:
            return func(*args, **kwargs)
        except (BdbQuit, KeyboardInterrupt):
            raise
        except Exception:
            msg = "Uncaught exception"
            logger.exception(msg)
            if with_debugger:
                import pdb  # noqa: T100
                import traceback

                traceback.print_exc()
                pdb.post_mortem()
            else:
                raise

    return wrapped

importers

import_module_from_info(module_info: ModuleInfo) -> ModuleType

Import a module from a ModuleInfo object.

Source code in src/rra_tools/cli_tools/importers.py 
 5
 6
 7
 8
 9
10
def import_module_from_info(module_info: ModuleInfo) -> ModuleType:
    """Import a module from a ModuleInfo object."""
    finder = module_info.module_finder
    spec = finder.find_spec(module_info.name)  # type: ignore[call-arg]
    module = spec.loader.load_module(module_info.name)  # type: ignore[union-attr]
    return module  # noqa: RET504

options

convert_choice(value: str, choices: Collection[str]) -> list[str]

Convert a choice to a list of choices, handling the special 'All' choice.

Parameters

value The choice to convert. choices The set of choices to choose from.

Returns

list[str] The list of choices.

Source code in src/rra_tools/cli_tools/options.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def convert_choice(value: str, choices: Collection[str]) -> list[str]:
    """Convert a choice to a list of choices, handling the special 'All' choice.

    Parameters
    ----------
    value
        The choice to convert.
    choices
        The set of choices to choose from.

    Returns
    -------
    list[str]
        The list of choices.
    """
    if value == RUN_ALL:
        return list(choices)
    elif value in choices:
        return [value]
    else:
        msg = f"Invalid choice: {value}. Must be one of {choices} or {RUN_ALL}."
        raise ValueError(msg)

process_choices(allow_all: bool, choices: Collection[str] | None) -> tuple[click.ParamType, str | None, bool]

Support function for creating options with choices.

A common pattern in RRA pipelines is to build CLIs that admit a choice of a specific set of values or a special value that represents all possible values. This function provides a way to handle this pattern in a consistent way.

There are four possible cases: 1. No choices are provided and RUN_ALL is allowed. This is useful when the set of choices is not known ahead of time, or is contingent on another option. For example, if there is a task that depends on location and year, but the years available depend on the location. The user might want to run a single year for a location (which they'll have to know ahead of time); or all years for a location, which would be the subset of years available for that location; or all years for all locations, which could be a different subset of years for each included location. 2. Choices are provided and RUN_ALL is allowed. This is useful when the set of choices is known ahead of time, but the user might want to run all of them. 3. No choices are provided and RUN_ALL is not allowed. This is useful when the set of choices is not known ahead of time, but the user must provide a value. 4. Choices are provided and RUN_ALL is not allowed. This is useful when the set of choices is known ahead of time and the user must provide a value.

Parameters

allow_all Whether to allow the special value RUN_ALL. choices The set of choices to allow.

Returns

tuple[click.ParamType, str | None, bool] The option type, default value, and whether to show the default.

Source code in src/rra_tools/cli_tools/options.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def process_choices(
    allow_all: bool,  # noqa: FBT001
    choices: Collection[str] | None,
) -> tuple[click.ParamType, str | None, bool]:
    """Support function for creating options with choices.

    A common pattern in RRA pipelines is to build CLIs that admit a choice
    of a specific set of values or a special value that represents all
    possible values. This function provides a way to handle this pattern
    in a consistent way.

    There are four possible cases:
    1. No choices are provided and RUN_ALL is allowed. This is useful when the
        set of choices is not known ahead of time, or is contingent on another
        option. For example, if there is a task that depends on location and year,
        but the years available depend on the location. The user might want to
        run a single year for a location (which they'll have to know ahead of time);
        or all years for a location, which would be the subset of years available
        for that location; or all years for all locations, which could be a different
        subset of years for each included location.
    2. Choices are provided and RUN_ALL is allowed. This is useful when the set of
        choices is known ahead of time, but the user might want to run all of them.
    3. No choices are provided and RUN_ALL is not allowed. This is useful when the
        set of choices is not known ahead of time, but the user must provide a value.
    4. Choices are provided and RUN_ALL is not allowed. This is useful when the set of
        choices is known ahead of time and the user must provide a value.

    Parameters
    ----------
    allow_all
        Whether to allow the special value RUN_ALL.
    choices
        The set of choices to allow.

    Returns
    -------
    tuple[click.ParamType, str | None, bool]
        The option type, default value, and whether to show the default.
    """

    if choices is None:
        option_type: click.ParamType = click.STRING
        default = RUN_ALL if allow_all else None
    else:
        choices = list(choices)
        if allow_all:
            choices.append(RUN_ALL)
            default = RUN_ALL
        else:
            default = None
        option_type = click.Choice(choices)
    show_default = default is not None
    return option_type, default, show_default

with_choice(name: str, short_name: str | None = None, *, allow_all: bool = True, choices: Collection[str] | None = None, convert: bool | None = None, **kwargs: Any) -> Callable[[Callable[P, T]], Callable[P, T]]

Create an option with a set of choices.

Parameters

name The name of the option. short_name An optional short name for the option. allow_all Whether to allow the special value "ALL", which represents all choices. choices The set of choices to allow. convert Whether to convert the provided argument to a list, resolving the special value "ALL" to all choices.

Source code in src/rra_tools/cli_tools/options.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def with_choice[**P, T](
    name: str,
    short_name: str | None = None,
    *,
    allow_all: bool = True,
    choices: Collection[str] | None = None,
    convert: bool | None = None,
    **kwargs: Any,
) -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Create an option with a set of choices.

    Parameters
    ----------
    name
        The name of the option.
    short_name
        An optional short name for the option.
    allow_all
        Whether to allow the special value "ALL", which represents all choices.
    choices
        The set of choices to allow.
    convert
        Whether to convert the provided argument to a list, resolving the special
        value "ALL" to all choices.

    """

    names = [f"--{name.replace('_', '-')}"]
    if short_name is not None:
        if len(short_name) != 1:
            msg = "Short names must be a single character."
            raise ValueError(msg)
        names.append(f"-{short_name}")
    option_type, default, show_default = process_choices(allow_all, choices)

    if choices and convert is None:
        convert = allow_all

    if convert:
        if not allow_all:
            msg = "Conversion is only supported when allow_all is True."
            raise ValueError(msg)
        if choices is None:
            msg = "Conversion is only supported when choices are provided."
            raise ValueError(msg)

        if "callback" in kwargs:
            old_callback = kwargs.pop("callback")

            def _callback(
                ctx: click.Context,
                param: click.Parameter,
                value: str,
            ) -> list[str]:
                value = old_callback(ctx, param, value)
                return convert_choice(value, choices)
        else:

            def _callback(
                ctx: click.Context,  # noqa: ARG001
                param: click.Parameter,  # noqa: ARG001
                value: str,
            ) -> list[str]:
                return convert_choice(value, choices)

        kwargs["callback"] = _callback

    return click.option(
        *names,
        type=option_type,
        default=default,
        show_default=show_default,
        **kwargs,
    )