Skip to content

Dynamic typer

DynamicTyper

Source code in src/snk_cli/dynamic_typer.py 
 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
 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
 87
 88
 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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
class DynamicTyper:
    app: typer.Typer

    def __init__(self):
        self.app = typer.Typer()

    def __call__(self):
        """
        Invoke the CLI.

        Side Effects:
          Invokes the CLI.

        Examples:
          >>> CLI(Path('/path/to/workflow'))()
        """
        self.app()

    def _set_app(self):
        """
        Set the app attribute.

        Side Effects:
          Sets the app attribute to a Typer object.
        """
        if not hasattr(self, "app"):
          self.app = typer.Typer()

    def register_default_command(self, command: Callable, **command_kwargs) -> None:
        """
        Register a default command to the CLI.

        Args:
          command (Callable): The command to register.

        Side Effects:
          Registers the command to the CLI.

        Examples:
          >>> CLI.register_default_command(my_command)
        """
        from makefun import with_signature
        from inspect import signature, Parameter

        command_signature = signature(command)
        params = list(command_signature.parameters.values())
        has_ctx = any([p.name == "ctx" for p in params])
        if not has_ctx:
            params.insert(
                0,
                Parameter(
                    "ctx",
                    kind=Parameter.POSITIONAL_OR_KEYWORD,
                    annotation=typer.Context,
                ),
            )
            command_signature = command_signature.replace(parameters=params)

        @with_signature(command_signature)
        def wrapper(ctx: typer.Context, *args, **kwargs):
            if ctx.invoked_subcommand is None:
                if has_ctx:
                    return command(ctx, *args, **kwargs)
                return command(*args, **kwargs)

        self.register_callback(wrapper, invoke_without_command=True, **command_kwargs)

    def register_command(
        self, command: Callable, dynamic_options=None, **command_kwargs
    ) -> None:
        """
        Register a command to the CLI.

        Args:
          command (Callable): The command to register.
          dynamic_options (List[Option], optional): A list of dynamic options to add to the command.

        Side Effects:
          Registers the command to the CLI.

        Examples:
          >>> CLI.register_command(my_command)
          >>> CLI.register_command(my_command, dynamic_options=[option1, option2])
        """
        self._set_app()
        if dynamic_options is not None:
            command = self.add_dynamic_options(command, dynamic_options)
        if isinstance(command, DynamicTyper):
            self.app.registered_commands.extend(command.app.registered_commands)
        else:
            self.app.command(**command_kwargs)(command)

    def register_callback(self, command: Callable, **command_kwargs) -> None:
        """
        Register a callback to the CLI.

        Args:
          command (Callable): The callback to register.

        Side Effects:
          Registers the callback to the CLI.

        Examples:
          >>> CLI.register_callback(my_callback)
        """
        self._set_app()
        self.app.callback(**command_kwargs)(command)

    def register_group(self, group: "DynamicTyper", **command_kwargs) -> None:
        """
        Register a subcommand group to the CLI.

        Args:
          group (DynamicTyper): The subcommand group to register.

        Side Effects:
          Registers the subcommand group to the CLI.

        Examples:
          >>> CLI.register_group(my_group)
        """
        self._set_app()
        self.app.add_typer(group.app, **command_kwargs)

    def _create_cli_parameter(self, option: Option):
        """
        Creates a parameter for a CLI option.

        Args:
          option (Option): An Option object containing the option's name, type, required status, default value, and help message.

        Returns:
          Parameter: A parameter object for the CLI option.

        Examples:
          >>> option = Option(name='foo', type='int', required=True, default=0, help='A number')
          >>> create_cli_parameter(option)
          Parameter('foo', kind=Parameter.POSITIONAL_OR_KEYWORD, default=typer.Option(..., help='[CONFIG] A number'), annotation=int)
        """
        return Parameter(
            option.name,
            kind=Parameter.POSITIONAL_OR_KEYWORD,
            default=typer.Option(
                ... if option.required else option.default,
                *[option.flag, option.short_flag] if option.short else [],
                help=f"{option.help}",
                rich_help_panel="Workflow Configuration",
                hidden=option.hidden,
            ),
            annotation=option.type,
        )

    def check_if_option_passed_via_command_line(self, option: Option):
        """
        Check if an option is passed via the command line.

        Args:
          option (Option): An Option object containing the option's name, type, required status, default value, and help message.

        Returns:
          bool: Whether the option is passed via the command line.
        """
        if option.flag in sys.argv:
            return True
        elif option.type is bool and f"--no-{option.flag[2:]}" in sys.argv:
            # Check for boolean flags like --foo/--no-foo
            return True
        elif option.short and option.short_flag in sys.argv:
            return True
        return False

    def add_dynamic_options(self, func: Callable, options: List[Option]):
        """
        Function to add dynamic options to a command.

        Args:
          func (Callable): The command to which the dynamic options should be added.
          options (List[Option]): A list of Option objects containing the options to add.

        Returns:
          Callable: A function with the dynamic options added.

        Examples:
          >>> my_func = add_dynamic_options_to_function(my_func, [option1, option2])
          >>> my_func
        """
        func_sig = signature(func)
        params = list(func_sig.parameters.values())
        for op in options[::-1]:
            params.insert(1, self._create_cli_parameter(op))
        new_sig = func_sig.replace(parameters=params)

        @with_signature(func_signature=new_sig, func_name=func.__name__)
        def func_wrapper(*args, **kwargs):
            """
            Wraps a function with dynamic options.

            Args:
              *args: Variable length argument list.
              **kwargs: Arbitrary keyword arguments.

            Returns:
              Callable: A wrapped function with the dynamic options added.

            Notes:
              This function is used in the `add_dynamic_options_to_function` function.
            """
            flat_config = None

            if kwargs.get("configfile"):
                from snakemake import load_configfile
                from .utils import flatten

                snakemake_config = load_configfile(kwargs["configfile"])
                flat_config = flatten(snakemake_config)

            for snk_cli_option in options:

                def add_option_to_args():
                    kwargs["ctx"].args.extend([f"--{snk_cli_option.name}", kwargs[snk_cli_option.name]])

                passed_via_command_line = self.check_if_option_passed_via_command_line(
                    snk_cli_option
                )

                if flat_config is None:
                    # If no config file is provided then all options should be added to the arguments
                    # later on we will check to see if they differ from any defaults
                    add_option_to_args()
                elif passed_via_command_line:
                    # If an option is passed via the command line if should override the default
                    add_option_to_args()
                elif flat_config and snk_cli_option.original_key not in flat_config:
                    # If a config file is provided and the snk_cli_option key isn't in it, 
                    # add the snk_cli_option to the arguments
                    add_option_to_args()

            kwargs = {
                k: v for k, v in kwargs.items() if k in func_sig.parameters.keys()
            }
            return func(*args, **kwargs)

        return func_wrapper

    def error(self, msg, exit=True):
        """
        Logs an error message (red) and exits (optional).

        Args:
          msg (str): The error message to log.
          exit (bool): Whether to exit after logging the error message.
        """
        typer.secho(msg, fg="red", err=True)
        if exit:
            raise typer.Exit(1)

    def success(self, msg):
        """
        Logs a success message (green).

        Args:
          msg (str): The success message to log.
        """
        typer.secho(msg, fg="green")

    def log(self, msg, color="yellow", stderr=True):
        """
        Logs a message (yellow).

        Args:
          msg (str): The message to log.
          color (str, optional): The color of the log message. Defaults to "yellow".
          stderr (bool, optional): Whether to log the message to stderr. Defaults to True.
        """
        typer.secho(msg, fg=color, err=stderr)

    def echo(self, msg):
        """
        Prints a message.

        Args:
          msg (str): The message to print.
        """
        typer.echo(msg)

__call__()

Invoke the CLI.

Side Effects

Invokes the CLI.

Examples:

>>> CLI(Path('/path/to/workflow'))()
Source code in src/snk_cli/dynamic_typer.py 
16
17
18
19
20
21
22
23
24
25
26
def __call__(self):
    """
    Invoke the CLI.

    Side Effects:
      Invokes the CLI.

    Examples:
      >>> CLI(Path('/path/to/workflow'))()
    """
    self.app()

add_dynamic_options(func, options)

Function to add dynamic options to a command.

Parameters:

Name Type Description Default
func Callable

The command to which the dynamic options should be added.

 required
options List[Option]

A list of Option objects containing the options to add.

 required

Returns:

Name Type Description
Callable

A function with the dynamic options added.

Examples:

>>> my_func = add_dynamic_options_to_function(my_func, [option1, option2])
>>> my_func
Source code in src/snk_cli/dynamic_typer.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
def add_dynamic_options(self, func: Callable, options: List[Option]):
    """
    Function to add dynamic options to a command.

    Args:
      func (Callable): The command to which the dynamic options should be added.
      options (List[Option]): A list of Option objects containing the options to add.

    Returns:
      Callable: A function with the dynamic options added.

    Examples:
      >>> my_func = add_dynamic_options_to_function(my_func, [option1, option2])
      >>> my_func
    """
    func_sig = signature(func)
    params = list(func_sig.parameters.values())
    for op in options[::-1]:
        params.insert(1, self._create_cli_parameter(op))
    new_sig = func_sig.replace(parameters=params)

    @with_signature(func_signature=new_sig, func_name=func.__name__)
    def func_wrapper(*args, **kwargs):
        """
        Wraps a function with dynamic options.

        Args:
          *args: Variable length argument list.
          **kwargs: Arbitrary keyword arguments.

        Returns:
          Callable: A wrapped function with the dynamic options added.

        Notes:
          This function is used in the `add_dynamic_options_to_function` function.
        """
        flat_config = None

        if kwargs.get("configfile"):
            from snakemake import load_configfile
            from .utils import flatten

            snakemake_config = load_configfile(kwargs["configfile"])
            flat_config = flatten(snakemake_config)

        for snk_cli_option in options:

            def add_option_to_args():
                kwargs["ctx"].args.extend([f"--{snk_cli_option.name}", kwargs[snk_cli_option.name]])

            passed_via_command_line = self.check_if_option_passed_via_command_line(
                snk_cli_option
            )

            if flat_config is None:
                # If no config file is provided then all options should be added to the arguments
                # later on we will check to see if they differ from any defaults
                add_option_to_args()
            elif passed_via_command_line:
                # If an option is passed via the command line if should override the default
                add_option_to_args()
            elif flat_config and snk_cli_option.original_key not in flat_config:
                # If a config file is provided and the snk_cli_option key isn't in it, 
                # add the snk_cli_option to the arguments
                add_option_to_args()

        kwargs = {
            k: v for k, v in kwargs.items() if k in func_sig.parameters.keys()
        }
        return func(*args, **kwargs)

    return func_wrapper

check_if_option_passed_via_command_line(option)

Check if an option is passed via the command line.

Parameters:

Name Type Description Default
option Option

An Option object containing the option's name, type, required status, default value, and help message.

 required

Returns:

Name Type Description
bool

Whether the option is passed via the command line.
Source code in src/snk_cli/dynamic_typer.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def check_if_option_passed_via_command_line(self, option: Option):
    """
    Check if an option is passed via the command line.

    Args:
      option (Option): An Option object containing the option's name, type, required status, default value, and help message.

    Returns:
      bool: Whether the option is passed via the command line.
    """
    if option.flag in sys.argv:
        return True
    elif option.type is bool and f"--no-{option.flag[2:]}" in sys.argv:
        # Check for boolean flags like --foo/--no-foo
        return True
    elif option.short and option.short_flag in sys.argv:
        return True
    return False

echo(msg)

Prints a message.

Parameters:

Name Type Description Default
msg str

The message to print.

 required
Source code in src/snk_cli/dynamic_typer.py 
286
287
288
289
290
291
292
293
def echo(self, msg):
    """
    Prints a message.

    Args:
      msg (str): The message to print.
    """
    typer.echo(msg)

error(msg, exit=True)

Logs an error message (red) and exits (optional).

Parameters:

Name Type Description Default
msg str

The error message to log.

 required
exit bool

Whether to exit after logging the error message.

 True
Source code in src/snk_cli/dynamic_typer.py 
254
255
256
257
258
259
260
261
262
263
264
def error(self, msg, exit=True):
    """
    Logs an error message (red) and exits (optional).

    Args:
      msg (str): The error message to log.
      exit (bool): Whether to exit after logging the error message.
    """
    typer.secho(msg, fg="red", err=True)
    if exit:
        raise typer.Exit(1)

log(msg, color='yellow', stderr=True)

Logs a message (yellow).

Parameters:

Name Type Description Default
msg str

The message to log.

 required
color str

The color of the log message. Defaults to "yellow".

 'yellow'
stderr bool

Whether to log the message to stderr. Defaults to True.

 True
Source code in src/snk_cli/dynamic_typer.py 
275
276
277
278
279
280
281
282
283
284
def log(self, msg, color="yellow", stderr=True):
    """
    Logs a message (yellow).

    Args:
      msg (str): The message to log.
      color (str, optional): The color of the log message. Defaults to "yellow".
      stderr (bool, optional): Whether to log the message to stderr. Defaults to True.
    """
    typer.secho(msg, fg=color, err=stderr)

register_callback(command, **command_kwargs)

Register a callback to the CLI.

Parameters:

Name Type Description Default
command Callable

The callback to register.

 required
Side Effects

Registers the callback to the CLI.

Examples:

>>> CLI.register_callback(my_callback)
Source code in src/snk_cli/dynamic_typer.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def register_callback(self, command: Callable, **command_kwargs) -> None:
    """
    Register a callback to the CLI.

    Args:
      command (Callable): The callback to register.

    Side Effects:
      Registers the callback to the CLI.

    Examples:
      >>> CLI.register_callback(my_callback)
    """
    self._set_app()
    self.app.callback(**command_kwargs)(command)

register_command(command, dynamic_options=None, **command_kwargs)

Register a command to the CLI.

Parameters:

Name Type Description Default
command Callable

The command to register.

 required
dynamic_options List[Option]

A list of dynamic options to add to the command.

 None
Side Effects

Registers the command to the CLI.

Examples:

>>> CLI.register_command(my_command)
>>> CLI.register_command(my_command, dynamic_options=[option1, option2])
Source code in src/snk_cli/dynamic_typer.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def register_command(
    self, command: Callable, dynamic_options=None, **command_kwargs
) -> None:
    """
    Register a command to the CLI.

    Args:
      command (Callable): The command to register.
      dynamic_options (List[Option], optional): A list of dynamic options to add to the command.

    Side Effects:
      Registers the command to the CLI.

    Examples:
      >>> CLI.register_command(my_command)
      >>> CLI.register_command(my_command, dynamic_options=[option1, option2])
    """
    self._set_app()
    if dynamic_options is not None:
        command = self.add_dynamic_options(command, dynamic_options)
    if isinstance(command, DynamicTyper):
        self.app.registered_commands.extend(command.app.registered_commands)
    else:
        self.app.command(**command_kwargs)(command)

register_default_command(command, **command_kwargs)

Register a default command to the CLI.

Parameters:

Name Type Description Default
command Callable

The command to register.

 required
Side Effects

Registers the command to the CLI.

Examples:

>>> CLI.register_default_command(my_command)
Source code in src/snk_cli/dynamic_typer.py 
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
def register_default_command(self, command: Callable, **command_kwargs) -> None:
    """
    Register a default command to the CLI.

    Args:
      command (Callable): The command to register.

    Side Effects:
      Registers the command to the CLI.

    Examples:
      >>> CLI.register_default_command(my_command)
    """
    from makefun import with_signature
    from inspect import signature, Parameter

    command_signature = signature(command)
    params = list(command_signature.parameters.values())
    has_ctx = any([p.name == "ctx" for p in params])
    if not has_ctx:
        params.insert(
            0,
            Parameter(
                "ctx",
                kind=Parameter.POSITIONAL_OR_KEYWORD,
                annotation=typer.Context,
            ),
        )
        command_signature = command_signature.replace(parameters=params)

    @with_signature(command_signature)
    def wrapper(ctx: typer.Context, *args, **kwargs):
        if ctx.invoked_subcommand is None:
            if has_ctx:
                return command(ctx, *args, **kwargs)
            return command(*args, **kwargs)

    self.register_callback(wrapper, invoke_without_command=True, **command_kwargs)

register_group(group, **command_kwargs)

Register a subcommand group to the CLI.

Parameters:

Name Type Description Default
group DynamicTyper

The subcommand group to register.

 required
Side Effects

Registers the subcommand group to the CLI.

Examples:

>>> CLI.register_group(my_group)
Source code in src/snk_cli/dynamic_typer.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def register_group(self, group: "DynamicTyper", **command_kwargs) -> None:
    """
    Register a subcommand group to the CLI.

    Args:
      group (DynamicTyper): The subcommand group to register.

    Side Effects:
      Registers the subcommand group to the CLI.

    Examples:
      >>> CLI.register_group(my_group)
    """
    self._set_app()
    self.app.add_typer(group.app, **command_kwargs)

success(msg)

Logs a success message (green).

Parameters:

Name Type Description Default
msg str

The success message to log.

 required
Source code in src/snk_cli/dynamic_typer.py 
266
267
268
269
270
271
272
273
def success(self, msg):
    """
    Logs a success message (green).

    Args:
      msg (str): The success message to log.
    """
    typer.secho(msg, fg="green")