#66 Standard CLI options builder

doc/changes/unreleased.md

@@ -1 +1,5 @@
 # Unreleased
+
+## Features
+
+* #66: Implement a standard CLI command builder.

exasol/python_extension_common/cli/std_options.py

@@ -0,0 +1,238 @@
+from typing import Any, no_type_check
+import os
+import re
+from enum import Flag, Enum, auto
+import click
+
+
+class ParameterFormatters:
+    """
+    Class facilitating customization of the cli.
+
+    The idea is that some of the cli parameters can be programmatically customized based
+    on values of other parameters and externally supplied formatters. For example a specialized
+    version of the cli may want to provide its own url. Furthermore, this url will depend on
+    the user supplied parameter called "version". The solution is to set a formatter for the
+    url, for instance "http://my_stuff/{version}/my_data". If the user specifies non-empty version
+    parameter the url will be fully formed.
+
+    A formatter may include more than one parameter. In the previous example the url could,
+    for instance, also include a username: "http://my_stuff/{version}/{user}/my_data".
+
+    Note that customized parameters can only be updated in a callback function. There is no
+    way to inject them directly into the cli. Also, the current implementation doesn't perform
+    the update if the value of the parameter dressed with the callback is None.
+    """
+    def __init__(self):
+        self._formatters = {}
+
+    def __call__(self, ctx: click.Context, param: click.Parameter, value: Any | None) -> Any | None:
+
+        def update_parameter(parameter_name: str, formatter: str) -> None:
+            param_formatter = ctx.params.get(parameter_name, formatter)
+            if param_formatter:
+                # Enclose in double curly brackets all other parameters in the formatting string,
+                # to avoid the missing parameters' error. Below is an example of a formatter string
+                # before and after applying the regex, assuming the current parameter is 'version'.
+                # 'something-with-{version}/tailored-for-{user}' => 'something-with-{version}/tailored-for-{{user}}'
+                # We were looking for all occurrences of a pattern '{some_name}', where some_name is not version.
+                pattern = r'\{(?!' + (param.name or '') + r'\})\w+\}'
+                param_formatter = re.sub(pattern, lambda m: f'{{{m.group(0)}}}', param_formatter)
+                kwargs = {param.name: value}
+                ctx.params[parameter_name] = param_formatter.format(**kwargs)
+
+        if value is not None:
+            for prm_name, prm_formatter in self._formatters.items():
+                update_parameter(prm_name, prm_formatter)
+
+        return value
+
+    def set_formatter(self, custom_parameter_name: str, formatter: str) -> None:
+        """ Sets a formatter for a customizable parameter. """
+        self._formatters[custom_parameter_name] = formatter
+
+    def clear_formatters(self):
+        """ Deletes all formatters, mainly for testing purposes. """
+        self._formatters.clear()
+
+
+# This text will be displayed instead of the actual value for a "secret" option.
+SECRET_DISPLAY = '***'
+
+
+def secret_callback(ctx: click.Context, param: click.Option, value: Any):
+    """
+    Here we try to get the secret option value from an environment variable.
+    The reason for doing this in the callback instead of using a callable default is
+    that we don't want the default to be displayed in the prompt. There seems to
+    be no way of altering this behaviour.
+    """
+    if value == SECRET_DISPLAY:
+        envar_name = param.opts[0][2:].upper()
+        return os.environ.get(envar_name)
+    return value
+
+
+class StdTags(Flag):
+    DB = auto()
+    BFS = auto()
+    ONPREM = auto()
+    SAAS = auto()
+    SLC = auto()
+
+
+class StdParams(Enum):
+    """
+    Standard option keys.
+    """
+    bucketfs_name = (StdTags.BFS | StdTags.ONPREM, auto())
+    bucketfs_host = (StdTags.BFS | StdTags.ONPREM, auto())
+    bucketfs_port = (StdTags.BFS | StdTags.ONPREM, auto())
+    bucketfs_use_https = (StdTags.BFS | StdTags.ONPREM, auto())
+    bucketfs_user = (StdTags.BFS | StdTags.ONPREM, auto())
+    bucketfs_password = (StdTags.BFS | StdTags.ONPREM, auto())
+    bucket = (StdTags.BFS | StdTags.ONPREM, auto())
+    saas_url = (StdTags.DB | StdTags.BFS | StdTags.SAAS, auto())
+    saas_account_id = (StdTags.DB | StdTags.BFS | StdTags.SAAS, auto())
+    saas_database_id = (StdTags.DB | StdTags.BFS | StdTags.SAAS, auto())
+    saas_database_name = (StdTags.DB | StdTags.BFS | StdTags.SAAS, auto())
+    saas_token = (StdTags.DB | StdTags.BFS | StdTags.SAAS, auto())
+    path_in_bucket = (StdTags.BFS | StdTags.ONPREM | StdTags.SAAS, auto())
+    container_file = (StdTags.SLC, auto())
+    version = (StdTags.SLC, auto())
+    dsn = (StdTags.DB | StdTags.ONPREM, auto())
+    db_user = (StdTags.DB | StdTags.ONPREM, auto())
+    db_password = (StdTags.DB | StdTags.ONPREM, auto())
+    language_alias = (StdTags.SLC, auto())
+    schema = (StdTags.DB | StdTags.ONPREM | StdTags.SAAS, auto())
+    ssl_cert_path = (StdTags.DB | StdTags.ONPREM, auto())
+    ssl_client_cert_path = (StdTags.DB | StdTags.ONPREM, auto())
+    ssl_client_private_key = (StdTags.DB | StdTags.ONPREM, auto())
+    use_ssl_cert_validation = (StdTags.DB | StdTags.BFS | StdTags.ONPREM, auto())
+    upload_container = (StdTags.SLC, auto())
+    alter_system = (StdTags.SLC, auto())
+    allow_override = (StdTags.SLC, auto())
+    wait_for_completion = (StdTags.SLC, auto())
+
+    def __init__(self, tags: StdTags, value):
+        self.tags = tags
+
+
+"""
+Standard options defined in the form of key-value pairs, where key is the option's
+StaParam key and the value is a kwargs for creating the click.Options(...).
+"""
+_std_options = {
+    StdParams.bucketfs_name: {'type': str},
+    StdParams.bucketfs_host: {'type': str},
+    StdParams.bucketfs_port: {'type': int},
+    StdParams.bucketfs_use_https: {'type': bool, 'default': False},
+    StdParams.bucketfs_user: {'type': str},
+    StdParams.bucketfs_password: {'type': str, 'hide_input': True},
+    StdParams.bucket: {'type': str},
+    StdParams.saas_url: {'type': str, 'default': 'https://cloud.exasol.com'},
+    StdParams.saas_account_id: {'type': str, 'hide_input': True},
+    StdParams.saas_database_id: {'type': str, 'hide_input': True},
+    StdParams.saas_database_name: {'type': str},
+    StdParams.saas_token: {'type': str, 'hide_input': True},
+    StdParams.path_in_bucket: {'type': str},
+    StdParams.container_file: {'type': click.Path(exists=True, file_okay=True)},
+    StdParams.version: {'type': str, 'expose_value': False},
+    StdParams.dsn: {'type': str},
+    StdParams.db_user: {'type': str},
+    StdParams.db_password: {'type': str, 'hide_input': True},
+    StdParams.language_alias: {'type': str},
+    StdParams.schema: {'type': str, 'default': ''},
+    StdParams.ssl_cert_path: {'type': str, 'default': ''},
+    StdParams.ssl_client_cert_path: {'type': str, 'default': ''},
+    StdParams.ssl_client_private_key: {'type': str, 'default': ''},
+    StdParams.use_ssl_cert_validation: {'type': bool, 'default': True},
+    StdParams.upload_container: {'type': bool, 'default': True},
+    StdParams.alter_system: {'type': bool, 'default': True},
+    StdParams.allow_override: {'type': bool, 'default': False},
+    StdParams.wait_for_completion: {'type': bool, 'default': True}
+}
+
+
+def make_option_secret(option_params: dict[str, Any], prompt: str) -> None:
+    """
+    Makes an option "secret" in the way that its input is not leaked to the
+    terminal. The option can be either a standard or a user defined.
+
+    Parameters:
+    option_params:
+        Option properties.
+    prompt:
+        The prompt text for this option.
+    """
+    option_params['hide_input'] = True
+    option_params['prompt'] = prompt
+    option_params['prompt_required'] = False
+    option_params['default'] = SECRET_DISPLAY
+    option_params['callback'] = secret_callback
+
+
+def create_std_option(std_param: StdParams, **kwargs) -> click.Option:
+    """
+    Creates a Click option.
+
+    Parameters:
+    std_param:
+        The option's StdParam key.
+    kwargs:
+        The option properties.
+    """
+    option_name = std_param.name.replace('_', '-')
+    if kwargs.get('type') == bool:
+        param_decls = [f'--{option_name}/--no-{option_name}']
+    else:
+        param_decls = [f'--{option_name}']
+    if kwargs.get('hide_input', False):
+        make_option_secret(kwargs, prompt=std_param.name.replace('_', ' '))
+    return click.Option(param_decls, **kwargs)
+
+
+@no_type_check
+def select_std_options(tags: StdTags | list[StdTags] | str,
+                       exclude: StdParams | list[StdParams] | None = None,
+                       override: dict[StdParams, dict[str, Any]] | None = None,
+                       formatters: dict[StdParams, ParameterFormatters] | None = None
+                       ) -> list[click.Option]:
+    """
+    Selects all or a subset of the defined standard Click options.
+
+    Parameters:
+    tags:
+        A flag or a list of flags that define the option selection criteria. Each flag
+        is a combination of the StdTags. An option gets selected if it's StdParams.tags
+        property includes any of the provided flags.
+        If the tags is the string "all" all the standard options will be selected.
+    exclude:
+        An option or a list of options that should not to be included in the output even
+        though they match the tags criteria.
+    override:
+        A dictionary of standard options with overridden properties
+    formatters:
+    """
+    if not isinstance(tags, list) and not isinstance(tags, str):
+        tags = [tags]
+    if exclude is None:
+        exclude = []
+    elif not isinstance(exclude, list):
+        exclude = [exclude]
+    override = override or {}
+    formatters = formatters or {}
+
+    def options_filter(std_param: StdParams) -> bool:
+        return any(tag in std_param.tags for tag in tags) and std_param not in exclude
+
+    def option_params(std_param: StdParams) -> dict[str, Any]:
+        return override[std_param] if std_param in override else _std_options[std_param]
+
+    if tags == 'all':
+        filtered_params = _std_options
+    else:
+        filtered_params = filter(options_filter, _std_options)
+    return [create_std_option(std_param, **option_params(std_param),
+                              callback=formatters.get(std_param))
+            for std_param in filtered_params]

test/unit/cli/test_std_options.py

@@ -0,0 +1,122 @@
+import os
+import click
+from click.testing import CliRunner
+from exasol.python_extension_common.cli.std_options import (
+    ParameterFormatters,
+    SECRET_DISPLAY,
+    StdTags,
+    StdParams,
+    create_std_option,
+    select_std_options
+)
+
+
+def test_parameter_formatters_1param():
+    container_url_param = 'container_url'
+    cmd = click.Command('a_command')
+    ctx = click.Context(cmd)
+    opt = click.Option(['--version'])
+    formatters = ParameterFormatters()
+    formatters.set_formatter(container_url_param, 'http://my_server/{version}/my_stuff')
+    formatters(ctx, opt, '1.3.2')
+    assert ctx.params[container_url_param] == 'http://my_server/1.3.2/my_stuff'
+
+
+def test_parameter_formatters_2params():
+    container_url_param = 'container_url'
+    container_name_param = 'container_name'
+    cmd = click.Command('a_command')
+    ctx = click.Context(cmd)
+    opt1 = click.Option(['--version'])
+    opt2 = click.Option(['--user'])
+    formatters = ParameterFormatters()
+    formatters.set_formatter(container_url_param, 'http://my_server/{version}/{user}/my_stuff')
+    formatters.set_formatter(container_name_param, 'downloaded-{version}')
+    formatters(ctx, opt1, '1.3.2')
+    formatters(ctx, opt2, 'cezar')
+    assert ctx.params[container_url_param] == 'http://my_server/1.3.2/cezar/my_stuff'
+    assert ctx.params[container_name_param] == 'downloaded-1.3.2'
+
+
+def test_create_std_option():
+    opt = create_std_option(StdParams.bucketfs_name, type=str)
+    assert opt.name == StdParams.bucketfs_name.name
+
+
+def test_create_std_option_bool():
+    opt = create_std_option(StdParams.allow_override, type=bool)
+    assert opt.name == StdParams.allow_override.name
+    assert '--no-allow-override' in opt.secondary_opts
+
+
+def test_create_std_option_secret():
+    opt = create_std_option(StdParams.db_password, type=str, hide_input=True)
+    assert opt.hide_input
+    assert not opt.prompt_required
+    assert opt.default == SECRET_DISPLAY
+
+
+def test_select_std_options():
+    for tag in StdTags:
+        opts = {opt.name for opt in select_std_options(tag)}
+        expected_opts = {std_param.name for std_param in StdParams if tag in std_param.tags}
+        assert opts == expected_opts
+
+
+def test_select_std_options_all():
+    opts = {opt.name for opt in select_std_options('all')}
+    expected_opts = {std_param.name for std_param in StdParams}
+    assert opts == expected_opts
+
+
+def test_select_std_options_restricted():
+    opts = {opt.name for opt in select_std_options(StdTags.BFS)}
+    opts_onprem = {opt.name for opt in select_std_options(StdTags.BFS | StdTags.ONPREM)}
+    assert opts_onprem
+    assert len(opts) > len(opts_onprem)
+    assert opts.intersection(opts_onprem) == opts_onprem
+
+
+def test_select_std_options_multi_tags():
+    opts = {opt.name for opt in select_std_options([StdTags.BFS, StdTags.SLC])}
+    expected_opts_bfs = {std_param.name for std_param in StdParams
+                         if StdTags.BFS in std_param.tags}
+    expected_opts_slc = {std_param.name for std_param in StdParams
+                         if StdTags.SLC in std_param.tags}
+    expected_opts = expected_opts_bfs.union(expected_opts_slc)
+    assert opts == expected_opts
+
+
+def test_select_std_options_with_exclude():
+    opts = [opt.name for opt in select_std_options(StdTags.SLC,
+                                                   exclude=StdParams.language_alias)]
+    assert StdParams.language_alias.name not in opts
+
+
+def test_select_std_options_with_override():
+    opts = {opt.name: opt for opt in select_std_options(
+        StdTags.SLC, override={StdParams.alter_system: {'type': bool, 'default': False}})}
+    assert not opts[StdParams.alter_system.name].default
+
+
+def test_hidden_opt_with_envar():
+    """
+    This test checks the mechanism of providing a value of a confidential parameter
+    via an environment variable.
+    """
+    std_param = StdParams.db_password
+    envar_name = std_param.name.upper()
+    param_value = 'my_password'
+
+    def func(**kwargs):
+        assert std_param.name in kwargs
+        assert kwargs[std_param.name] == param_value
+
+    opt = create_std_option(std_param, type=str, hide_input=True)
+    cmd = click.Command('do_something', params=[opt], callback=func)
+    runner = CliRunner()
+    os.environ[envar_name] = param_value
+    try:
+        runner.invoke(cmd)
+    finally:
+        os.environ.pop(envar_name)