refactor: Use inheritance to construct plugin CLI

docs/guides/custom-clis.md

@@ -0,0 +1,38 @@
+# Custom CLIs
+
+## Overview
+
+By default, packages created with the Singer SDK will have a single command, e.g. `tap-my-source`, which will run the application in a Singer-compatible way. However, you may want to add additional commands to your package. For example, you may want to add a command to initialize the database or platform with certain attributes required by the application to run properly.
+
+## Adding a custom command
+
+To add a custom command, you will need to add a new method to your plugin class that returns an instance of [`click.Command`](https://click.palletsprojects.com/en/8.1.x/api/#commands) (or a subclass of it) and decorate it with the `singer_sdk.cli.plugin_cli` decorator. Then you will need to add the command to the `[tool.poetry.scripts]` section of your `pyproject.toml` file.
+
+```python
+# tap_shortcut/tap.py
+
+class ShortcutTap(Tap):
+    """Shortcut tap class."""
+
+    @plugin_cli
+    def update_schema(cls) -> click.Command:
+        """Update the OpenAPI schema for this tap."""
+        @click.command()
+        def update():
+            response = requests.get(
+                "https://developer.shortcut.com/api/rest/v3/shortcut.swagger.json",
+                timeout=5,
+            )
+            with Path("tap_shortcut/openapi.json").open("w") as f:
+                f.write(response.text)
+
+        return update
+```
+
+```toml
+# pyproject.toml
+
+[tool.poetry.scripts]
+tap-shortcut = "tap_shortcut.tap:ShortcutTap.cli"
+tap-shortcut-update-schema = "tap_shortcut.tap:ShortcutTap.update_schema"
+```

docs/guides/index.md

@@ -7,4 +7,5 @@ The following pages contain useful information for developers building on top of
 
 porting
 pagination-classes
+custom-clis
 ```

pyproject.toml

@@ -306,6 +306,11 @@ parametrize-names-type = "csv"
 known-first-party = ["singer_sdk", "samples", "tests"]
 required-imports = ["from __future__ import annotations"]
 
+[tool.ruff.pep8-naming]
+classmethod-decorators = [
+    "singer_sdk.cli.plugin_cli",
+]
+
 [tool.ruff.pydocstyle]
 convention = "google"
 

singer_sdk/cli/__init__.py

@@ -1,3 +1,35 @@
 """Helpers for the tap, target and mapper CLIs."""
 
 from __future__ import annotations
+
+import typing as t
+
+if t.TYPE_CHECKING:
+    import click
+
+_T = t.TypeVar("_T")
+
+
+class plugin_cli:  # noqa: N801
+    """Decorator to create a plugin CLI."""
+
+    def __init__(self, method: t.Callable[..., click.Command]) -> None:
+        """Create a new plugin CLI.
+
+        Args:
+            method: The method to call to get the command.
+        """
+        self.method = method
+        self.name: str | None = None
+
+    def __get__(self, instance: _T, owner: type[_T]) -> click.Command:
+        """Get the command.
+
+        Args:
+            instance: The instance of the plugin.
+            owner: The plugin class.
+
+        Returns:
+            The CLI entrypoint.
+        """
+        return self.method(owner)

singer_sdk/configuration/_dict_config.py

@@ -85,13 +85,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():
             msg = (
                 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:
         msg = "Could not open file. Filepath not provided."
         raise RuntimeError(msg)

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."""
@@ -106,65 +102,54 @@ def map_batch_message(
         msg = "BATCH messages are not supported by mappers."
         raise NotImplementedError(msg)
 
-    @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_singer_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_singer_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
@@ -13,6 +14,7 @@
 from jsonschema import Draft7Validator
 
 from singer_sdk import about, metrics
+from singer_sdk.cli import plugin_cli
 from singer_sdk.configuration._dict_config import (
     merge_missing_config_jsonschema,
     parse_environment_config,
@@ -399,16 +401,99 @@ 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_singer_command(cls: type[PluginBase]) -> click.Command:
         """Handle command line execution.
 
         Returns:
             A callable CLI object.
         """
+        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,
+                ),
+            ],
+        )
 
-        @click.command()
-        def cli() -> None:
-            pass
+    @plugin_cli
+    def cli(cls) -> click.Command:
+        """Handle command line execution.
 
-        return cli
+        Returns:
+            A callable CLI object.
+        """
+        return cls.get_singer_command()