refactor: Use inheritance for plugins' CLI

singer_sdk/_python_types.py

@@ -0,0 +1,6 @@
+from __future__ import annotations
+
+import os
+from typing import Union
+
+_FilePath = Union[str, os.PathLike]

singer_sdk/cli/__init__.py

@@ -1,3 +1,7 @@
 """Helpers for the tap, target and mapper CLIs."""
 
 from __future__ import annotations
+
+from singer_sdk.cli.options import NestedOption
+
+__all__ = ["NestedOption"]

singer_sdk/cli/options.py

@@ -0,0 +1,70 @@
+"""Helpers for building the CLI for a Singer tap or target."""
+
+from __future__ import annotations
+
+from typing import Any, Mapping, Sequence
+
+import click
+
+
+class NestedOption(click.Option):
+    """A Click option that has suboptions."""
+
+    def __init__(
+        self,
+        *args: Any,
+        suboptions: Sequence[click.Option] | None = None,
+        **kwargs: Any,
+    ) -> None:
+        """Initialize the option.
+
+        Args:
+            *args: Positional arguments to pass to the parent class.
+            suboptions: A list of suboptions to be added to the context.
+            **kwargs: Keyword arguments to pass to the parent class.
+        """
+        self.suboptions = suboptions or []
+        super().__init__(*args, **kwargs)
+
+    def handle_parse_result(
+        self,
+        ctx: click.Context,
+        opts: Mapping[str, Any],
+        args: list[Any],
+    ) -> tuple[Any, list[str]]:
+        """Handle the parse result.
+
+        Args:
+            ctx: The Click context.
+            opts: The options.
+            args: The arguments.
+
+        Raises:
+            UsageError: If an option is used without the parent option.
+
+        Returns:
+            The parse result.
+        """
+        ctx.ensure_object(dict)
+        ctx.obj[self.name] = {}
+
+        if self.name in opts:
+            for option in self.suboptions:
+                if option.name:
+                    value = opts.get(option.name, option.get_default(ctx))
+                    ctx.obj[self.name][option.name] = value
+        else:
+            for option in self.suboptions:
+                if option.name in opts:
+                    errmsg = f"{option.opts[0]} is not allowed without {self.opts[0]}"
+                    raise click.UsageError(errmsg)
+
+        return super().handle_parse_result(ctx, opts, args)
+
+    def as_params(self) -> list[click.Option]:
+        """Return a list of options, including this one and its suboptions.
+
+        Returns:
+            List of options.
+        """
+        return [self, *self.suboptions]

singer_sdk/configuration/_dict_config.py

@@ -84,13 +84,15 @@ def merge_config_sources(
         A single configuration dictionary.
     """
     config: dict[str, Any] = {}
-    for config_path in inputs:
-        if config_path == "ENV":
+    for config_input in inputs:
+        if config_input == "ENV":
             env_config = parse_environment_config(config_schema, prefix=env_prefix)
             config.update(env_config)
             continue
 
-        if not Path(config_path).is_file():
+        config_path = Path(config_input)
+
+        if not config_path.is_file():
             raise FileNotFoundError(
                 f"Could not locate config file at '{config_path}'."
                 "Please check that the file exists."

singer_sdk/helpers/_util.py

@@ -3,25 +3,29 @@
 from __future__ import annotations
 
 import json
-from pathlib import Path, PurePath
+from pathlib import Path
 from typing import Any, cast
 
 import pendulum
 
+from singer_sdk._python_types import _FilePath
 
-def read_json_file(path: PurePath | str) -> dict[str, Any]:
-    """Read json file, thowing an error if missing."""
+
+def read_json_file(path: _FilePath) -> dict[str, Any]:
+    """Read json file, throwing an error if missing."""
     if not path:
         raise RuntimeError("Could not open file. Filepath not provided.")
 
-    if not Path(path).exists():
-        msg = f"File at '{path}' was not found."
-        for template in [f"{path}.template"]:
+    path_obj = Path(path)
+
+    if not path_obj.exists():
+        msg = f"File at '{path!r}' was not found."
+        for template in [f"{path!r}.template"]:
             if Path(template).exists():
                 msg += f"\nFor more info, please see the sample template at: {template}"
         raise FileExistsError(msg)
 
-    return cast(dict, json.loads(Path(path).read_text()))
+    return cast(dict, json.loads(path_obj.read_text()))
 
 
 def utc_now() -> pendulum.DateTime:

singer_sdk/mapper_base.py

@@ -3,14 +3,11 @@
 from __future__ import annotations
 
 import abc
-from io import FileIO
-from typing import Callable, Iterable
+from typing import IO, Iterable
 
 import click
 
 import singer_sdk._singerlib as singer
-from singer_sdk.cli import common_options
-from singer_sdk.configuration._dict_config import merge_config_sources
 from singer_sdk.helpers._classproperty import classproperty
 from singer_sdk.helpers.capabilities import CapabilitiesEnum, PluginCapabilities
 from singer_sdk.io_base import SingerReader
@@ -108,68 +105,48 @@ def map_batch_message(
         """
         raise NotImplementedError("BATCH messages are not supported by mappers.")
 
-    @classproperty
-    def cli(cls) -> Callable:
+    # CLI handler
+
+    @classmethod
+    def invoke(  # type: ignore[override]
+        cls: type[InlineMapper],
+        config: tuple[str, ...] = (),
+        file_input: IO[str] | None = None,
+    ) -> None:
+        """Invoke the mapper.
+
+        Args:
+            config: Configuration file location or 'ENV' to use environment
+                variables. Accepts multiple inputs as a tuple.
+            file_input: Optional file to read input from.
+        """
+        cls.print_version(print_fn=cls.logger.info)
+        config_files, parse_env_config = cls.config_from_cli_args(*config)
+
+        mapper = cls(
+            config=config_files,
+            validate_config=True,
+            parse_env_config=parse_env_config,
+        )
+        mapper.listen(file_input)
+
+    @classmethod
+    def get_command(cls: type[InlineMapper]) -> click.Command:
         """Execute standard CLI handler for inline mappers.
 
         Returns:
-            A callable CLI object.
+            A click.Command object.
         """
-
-        @common_options.PLUGIN_VERSION
-        @common_options.PLUGIN_ABOUT
-        @common_options.PLUGIN_ABOUT_FORMAT
-        @common_options.PLUGIN_CONFIG
-        @common_options.PLUGIN_FILE_INPUT
-        @click.command(
-            help="Execute the Singer mapper.",
-            context_settings={"help_option_names": ["--help"]},
+        command = super().get_command()
+        command.help = "Execute the Singer mapper."
+        command.params.extend(
+            [
+                click.Option(
+                    ["--input", "file_input"],
+                    help="A path to read messages from instead of from standard in.",
+                    type=click.File("r"),
+                ),
+            ],
         )
-        def cli(
-            version: bool = False,
-            about: bool = False,
-            config: tuple[str, ...] = (),
-            format: str | None = None,
-            file_input: FileIO | None = None,
-        ) -> None:
-            """Handle command line execution.
-
-            Args:
-                version: Display the package version.
-                about: Display package metadata and settings.
-                format: Specify output style for `--about`.
-                config: Configuration file location or 'ENV' to use environment
-                    variables. Accepts multiple inputs as a tuple.
-                file_input: Specify a path to an input file to read messages from.
-                    Defaults to standard in if unspecified.
-            """
-            if version:
-                cls.print_version()
-                return
-
-            if not about:
-                cls.print_version(print_fn=cls.logger.info)
-
-            validate_config: bool = True
-            if about:
-                validate_config = False
-
-            cls.print_version(print_fn=cls.logger.info)
-
-            config_dict = merge_config_sources(
-                config,
-                cls.config_jsonschema,
-                cls._env_prefix,
-            )
-
-            mapper = cls(  # type: ignore  # Ignore 'type not callable'
-                config=config_dict,
-                validate_config=validate_config,
-            )
-
-            if about:
-                mapper.print_about(format)
-            else:
-                mapper.listen(file_input)
-
-        return cli
+
+        return command