#127: Refactored class ParameterFormatters and docstrings

doc/changes/unreleased.md

@@ -1,3 +1,7 @@
 # Unreleased
 
 ## Summary
+
+## Refactorings
+
+* #127: Refactored class `ParameterFormatters` and docstrings

exasol/python_extension_common/cli/_param.py

@@ -0,0 +1,13 @@
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass(frozen=True)
+class Param:
+    """
+    Only used internally to distinguish between source and destination
+    parameters.
+    """
+
+    name: str | None
+    value: Any | None

exasol/python_extension_common/cli/std_options.py

@@ -12,56 +12,94 @@
 
 import click
 
+from exasol.python_extension_common.cli._param import Param
+
 
 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
+    patterns, called "formatters".
+
+    Example: A specialized variant of the CLI may want to provide a custom URL
+    "http://prefix/{version}/suffix" depending on CLI parameter "version".  If
+    the user specifies version "1.2.3", then the default value for the URL
+    should be updated to "http://prefix/1.2.3/suffix".
+
+    The URL parameter in this example is called a _destination_ CLI parameter
+    while the version is called _source_.
 
-    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 destination parameter can depend on a single or multiple source
+    parameters.  In the previous example the URL could, for instance, also
+    include a username: "http://prefix/{version}/{user}/suffix".
 
-    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".
+    The Click API allows updating customized parameters only in a callback
+    function.  There is no way to inject them directly into the CLI, see the
+    docs, as ``click.Option`` inherits from ``click.Parameter``:
+    https://click.palletsprojects.com/en/stable/api/#click.Parameter.
 
-    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.
+    The current implementation updates the destination parameter only if the
+    value of the source parameter is not ``None``.
     """
 
     def __init__(self) -> None:
+        # Each key/value pair represents the name of a destination parameter
+        # to update, and its default value.
+        #
+        # The default value can contain placeholders to be replaced by the
+        # values of the other parameters, called "source parameters".
         self._formatters: dict[str, str] = {}
 
-    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)  # type: ignore
-
-        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 __call__(
+        self, ctx: click.Context, source_param: click.Parameter, value: Any | None
+    ) -> Any | None:
+        def update(source: Param, dest: Param) -> None:
+            if not dest.value:
+                return None
+            # Enclose in double curly brackets all other parameters in
+            # dest.value to avoid error "missing parameters".
+            #
+            # Below is an example of a formatter string before and after
+            # applying the regex, assuming the source parameter is 'version'.
+            #
+            # "something-with-{version}/tailored-for-{user}" =>
+            # "something-with-{version}/tailored-for-{{user}}"
+            #
+            # We were looking for all occurrences of a pattern "{xxx}", where
+            # xxx is not "version".
+            pattern = r"\{(?!" + (source.name or "") + r"\})\w+\}"
+            template = re.sub(pattern, lambda m: f"{{{m.group(0)}}}", dest.value)
+            kwargs = {source.name: source.value}
+            ctx.params[dest.name] = template.format(**kwargs)  # type: ignore
+
+        source = Param(source_param.name, value)
+        if source.value is not None:
+            for name, default in self._formatters.items():
+                value = ctx.params.get(name, default)
+                update(source, dest=Param(name, value))
+
+        return source.value
+
+    def set_formatter(self, param_name: str, default_value: str) -> None:
+        """
+        Adds the specified destination parameter to be updated.
+
+        Better arg names could be `destination_parameter` and `format_pattern`
+        but renaming the arguments would break the public interface.
+
+        Parameters:
+
+          param_name: Name of the destination parameter to be updated.
+
+          default_value: Pattern for the value to assign to the destination
+                parameter. The pattern may contain place holders to be
+                replaced by the values of other CLI parameters, called "source
+                parameters".
+        """
+        self._formatters[param_name] = default_value
 
     def clear_formatters(self):
-        """Deletes all formatters, mainly for testing purposes."""
+        """Deletes all destination parameters to be updated, mainly for testing purposes."""
         self._formatters.clear()
 
 
@@ -253,6 +291,16 @@ def select_std_options(
     override:
         A dictionary of standard options with overridden properties
     formatters:
+        A dictionary, with each key being a source CLI parameter,
+        see docstring of class ParameterFormatters.
+
+        Each value is an instance of ParameterFormatters representing a single
+        or multiple destination parameters to be updated based on the source
+        parameter's value.
+
+        If a particular destination parameter D depends on multiple source
+        parameters S1, S2, ..., then the Click API will iterate through the
+        source parameters Si and update D multiple times.
     """
     if not isinstance(tags, list) and not isinstance(tags, str):
         tags = [tags]

exasol/python_extension_common/deployment/extract_validator.py

@@ -1,11 +1,11 @@
+from collections.abc import (
+    Callable,
+)
 from datetime import (
     datetime,
     timedelta,
 )
 from textwrap import dedent
-from typing import (
-    Callable,
-)
 
 import exasol.bucketfs as bfs  # type: ignore
 import pyexasol  # type: ignore

exasol/python_extension_common/deployment/language_container_builder.py

@@ -3,9 +3,9 @@
 import shutil
 import subprocess
 import tempfile
+from collections.abc import Callable
 from importlib import resources
 from pathlib import Path
-from typing import Callable
 
 from exasol.slc import api  # type: ignore
 from exasol.slc.models.compression_strategy import CompressionStrategy

exasol/python_extension_common/deployment/language_container_deployer.py

@@ -9,9 +9,6 @@
     PurePosixPath,
 )
 from textwrap import dedent
-from typing import (
-    Optional,
-)
 
 import exasol.bucketfs as bfs  # type: ignore
 import pyexasol  # type: ignore
@@ -32,9 +29,9 @@
 
 def get_websocket_sslopt(
     use_ssl_cert_validation: bool = True,
-    ssl_trusted_ca: Optional[str] = None,
-    ssl_client_certificate: Optional[str] = None,
-    ssl_private_key: Optional[str] = None,
+    ssl_trusted_ca: str | None = None,
+    ssl_client_certificate: str | None = None,
+    ssl_private_key: str | None = None,
 ) -> dict:
     """
     Returns a dictionary in the winsocket-client format
@@ -180,8 +177,8 @@ def _upload_path(self, bucket_file_path: str | None) -> bfs.path.PathLike:
 
     def run(
         self,
-        container_file: Optional[Path] = None,
-        bucket_file_path: Optional[str] = None,
+        container_file: Path | None = None,
+        bucket_file_path: str | None = None,
         alter_system: bool = True,
         allow_override: bool = False,
         wait_for_completion: bool = True,
@@ -245,9 +242,7 @@ def run(
             )
             print(message)
 
-    def upload_container(
-        self, container_file: Path, bucket_file_path: Optional[str] = None
-    ) -> None:
+    def upload_container(self, container_file: Path, bucket_file_path: str | None = None) -> None:
         """
         Upload the language container to the BucketFS.
 
@@ -376,27 +371,27 @@ def _check_if_requested_language_alias_already_exists(
     def create(
         cls,
         language_alias: str,
-        dsn: Optional[str] = None,
+        dsn: str | None = None,
         schema: str = "",
-        db_user: Optional[str] = None,
-        db_password: Optional[str] = None,
-        bucketfs_host: Optional[str] = None,
-        bucketfs_port: Optional[int] = None,
-        bucketfs_name: Optional[str] = None,
-        bucket: Optional[str] = None,
-        bucketfs_user: Optional[str] = None,
-        bucketfs_password: Optional[str] = None,
+        db_user: str | None = None,
+        db_password: str | None = None,
+        bucketfs_host: str | None = None,
+        bucketfs_port: int | None = None,
+        bucketfs_name: str | None = None,
+        bucket: str | None = None,
+        bucketfs_user: str | None = None,
+        bucketfs_password: str | None = None,
         bucketfs_use_https: bool = True,
-        saas_url: Optional[str] = None,
-        saas_account_id: Optional[str] = None,
-        saas_database_id: Optional[str] = None,
-        saas_database_name: Optional[str] = None,
-        saas_token: Optional[str] = None,
+        saas_url: str | None = None,
+        saas_account_id: str | None = None,
+        saas_database_id: str | None = None,
+        saas_database_name: str | None = None,
+        saas_token: str | None = None,
         path_in_bucket: str = "",
         use_ssl_cert_validation: bool = True,
-        ssl_trusted_ca: Optional[str] = None,
-        ssl_client_certificate: Optional[str] = None,
-        ssl_private_key: Optional[str] = None,
+        ssl_trusted_ca: str | None = None,
+        ssl_client_certificate: str | None = None,
+        ssl_private_key: str | None = None,
         deploy_timeout: timedelta = timedelta(minutes=10),
         display_progress: bool = False,
     ) -> "LanguageContainerDeployer":

exasol/python_extension_common/deployment/language_container_deployer_cli.py

@@ -6,7 +6,6 @@
 from pathlib import Path
 from typing import (
     Any,
-    Optional,
 )
 
 import click
@@ -28,6 +27,11 @@ class CustomizableParameters(Enum):
 
 
 class _ParameterFormatters:
+    # See language_container_deployer_main displaying a deprecation warning.
+    #
+    # Currently, there is only one other project still using this
+    # implementation, see
+    # https://github.com/exasol/advanced-analytics-framework/issues/333.
     """
     Class facilitating customization of the cli.
 
@@ -52,9 +56,7 @@ class _ParameterFormatters:
     def __init__(self):
         self._formatters = {}
 
-    def __call__(
-        self, ctx: click.Context, param: click.Parameter, value: Optional[Any]
-    ) -> Optional[Any]:
+    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)
@@ -226,8 +228,8 @@ def language_container_deployer_main(
     wait_for_completion: bool,
     deploy_timeout_minutes: int,
     display_progress: bool,
-    container_url: Optional[str] = None,
-    container_name: Optional[str] = None,
+    container_url: str | None = None,
+    container_name: str | None = None,
 ):
     warnings.warn(
         "language_container_deployer_main() function is deprecated and will be removed "

test/unit/cli/test_std_options.py

@@ -135,11 +135,11 @@ def func(**kwargs):
         assert kwargs[container_name_arg] == expected_name
         assert kwargs[container_url_arg] == expected_url
 
-    ver_formatter = ParameterFormatters()
-    ver_formatter.set_formatter(container_url_arg, url_format)
-    ver_formatter.set_formatter(container_name_arg, name_format)
+    ver_formatters = ParameterFormatters()
+    ver_formatters.set_formatter(container_url_arg, url_format)
+    ver_formatters.set_formatter(container_name_arg, name_format)
 
-    opts = select_std_options(StdTags.SLC, formatters={StdParams.version: ver_formatter})
+    opts = select_std_options(StdTags.SLC, formatters={StdParams.version: ver_formatters})
     cmd = click.Command("do_something", params=opts, callback=func)
     runner = CliRunner()
     runner.invoke(cmd, args=f"--version {version}", catch_exceptions=False, standalone_mode=False)