refactor: Use inheritance to construct plugin CLI

singer_sdk/configuration/_dict_config.py

@@ -84,13 +84,15 @@ def merge_config_sources(
         A single configuration dictionary.
     """
     config: dict[str, t.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

@@ -10,7 +10,7 @@
 
 
 def read_json_file(path: PurePath | str) -> dict[str, t.Any]:
-    """Read json file, thowing an error if missing."""
+    """Read json file, throwing an error if missing."""
     if not path:
         raise RuntimeError("Could not open file. Filepath not provided.")
 

singer_sdk/mapper_base.py

@@ -8,15 +8,11 @@
 import click
 
 import singer_sdk._singerlib as singer
-from singer_sdk.cli import common_options
 from singer_sdk.helpers._classproperty import classproperty
 from singer_sdk.helpers.capabilities import CapabilitiesEnum, PluginCapabilities
 from singer_sdk.io_base import SingerReader
 from singer_sdk.plugin_base import PluginBase
 
-if t.TYPE_CHECKING:
-    from io import FileIO
-
 
 class InlineMapper(PluginBase, SingerReader, metaclass=abc.ABCMeta):
     """Abstract base class for inline mappers."""
@@ -105,65 +101,54 @@ def map_batch_message(
         """
         raise NotImplementedError("BATCH messages are not supported by mappers.")
 
-    @classproperty
-    def cli(cls) -> t.Callable:  # noqa: N805
+    # CLI handler
+
+    @classmethod
+    def invoke(  # type: ignore[override]
+        cls: type[InlineMapper],
+        *,
+        about: bool = False,
+        about_format: str | None = None,
+        config: tuple[str, ...] = (),
+        file_input: t.IO[str] | None = None,
+    ) -> None:
+        """Invoke the mapper.
+
+        Args:
+            about: Display package metadata and settings.
+            about_format: Specify output style for `--about`.
+            config: Configuration file location or 'ENV' to use environment
+                variables. Accepts multiple inputs as a tuple.
+            file_input: Optional file to read input from.
+        """
+        super().invoke(about=about, about_format=about_format)
+        cls.print_version(print_fn=cls.logger.info)
+        config_files, parse_env_config = cls.config_from_cli_args(*config)
+
+        mapper = cls(
+            config=config_files,  # type: ignore[arg-type]
+            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, ...] = (),
-            about_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.
-                about_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_files, parse_env_config = cls.config_from_cli_args(*config)
-            mapper = cls(  # type: ignore[operator]
-                config=config_files or None,
-                validate_config=validate_config,
-                parse_env_config=parse_env_config,
-            )
-
-            if about:
-                mapper.print_about(about_format)
-            else:
-                mapper.listen(file_input)
-
-        return cli
+
+        return command

singer_sdk/plugin_base.py

@@ -5,6 +5,7 @@
 import abc
 import logging
 import os
+import sys
 import typing as t
 from pathlib import Path, PurePath
 from types import MappingProxyType
@@ -36,6 +37,25 @@
 JSONSchemaValidator = extend_validator_with_defaults(Draft7Validator)
 
 
+class PluginCLI:
+    """A descriptor for the plugin's CLI command."""
+
+    def __get__(self, instance: PluginBase, owner: type[PluginBase]) -> click.Command:
+        """Get the plugin's CLI command.
+
+        Args:
+            instance: The plugin instance.
+            owner: The plugin class.
+
+        Returns:
+            The plugin's CLI command.
+        """
+        if instance is None:
+            return owner.get_command()
+
+        return instance.get_command()
+
+
 class PluginBase(metaclass=abc.ABCMeta):
     """Abstract base class for taps."""
 
@@ -46,6 +66,8 @@ class PluginBase(metaclass=abc.ABCMeta):
 
     _config: dict
 
+    cli = PluginCLI()
+
     @classproperty
     def logger(cls) -> logging.Logger:  # noqa: N805
         """Get logger.
@@ -401,16 +423,90 @@ def config_from_cli_args(*args: str) -> tuple[list[Path], bool]:
 
         return config_files, parse_env_config
 
-    @classproperty
-    def cli(cls) -> t.Callable:  # noqa: N805
+    @classmethod
+    def invoke(
+        cls,
+        *,
+        about: bool = False,
+        about_format: str | None = None,
+        **kwargs: t.Any,  # noqa: ARG003
+    ) -> None:
+        """Invoke the plugin.
+
+        Args:
+            about: Display package metadata and settings.
+            about_format: Specify output style for `--about`.
+            kwargs: Plugin keyword arguments.
+        """
+        if about:
+            cls.print_about(about_format)
+            sys.exit(0)
+
+    @classmethod
+    def cb_version(
+        cls: type[PluginBase],
+        ctx: click.Context,
+        param: click.Option,  # noqa: ARG003
+        value: bool,  # noqa: FBT001
+    ) -> None:
+        """CLI callback to print the plugin version and exit.
+
+        Args:
+            ctx: Click context.
+            param: Click parameter.
+            value: Boolean indicating whether to print the version.
+        """
+        if not value:
+            return
+        cls.print_version(print_fn=click.echo)
+        ctx.exit()
+
+    @classmethod
+    def get_command(cls: type[PluginBase]) -> click.Command:
         """Handle command line execution.
 
         Returns:
             A callable CLI object.
         """
-
-        @click.command()
-        def cli() -> None:
-            pass
-
-        return cli
+        return click.Command(
+            name=cls.name,
+            callback=cls.invoke,
+            context_settings={"help_option_names": ["--help"]},
+            params=[
+                click.Option(
+                    ["--version"],
+                    is_flag=True,
+                    help="Display the package version.",
+                    is_eager=True,
+                    expose_value=False,
+                    callback=cls.cb_version,
+                ),
+                click.Option(
+                    ["--about"],
+                    help="Display package metadata and settings.",
+                    is_flag=True,
+                    is_eager=False,
+                    expose_value=True,
+                ),
+                click.Option(
+                    ["--format", "about_format"],
+                    help="Specify output style for --about",
+                    type=click.Choice(
+                        ["json", "markdown"],
+                        case_sensitive=False,
+                    ),
+                    default=None,
+                ),
+                click.Option(
+                    ["--config"],
+                    multiple=True,
+                    help=(
+                        "Configuration file location or 'ENV' to use environment "
+                        "variables."
+                    ),
+                    type=click.STRING,
+                    default=(),
+                    is_eager=True,
+                ),
+            ],
+        )