diff --git a/docs/guides/custom-clis.md b/docs/guides/custom-clis.md new file mode 100644 index 000000000..f17cabb5b --- /dev/null +++ b/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" +``` diff --git a/docs/guides/index.md b/docs/guides/index.md index c4f5f8d69..e86aa149c 100644 --- a/docs/guides/index.md +++ b/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 ``` diff --git a/pyproject.toml b/pyproject.toml index 0eeb20e1f..6f047d847 100644 --- a/pyproject.toml +++ b/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" diff --git a/singer_sdk/cli/__init__.py b/singer_sdk/cli/__init__.py index 76b58f748..cb0d72607 100644 --- a/singer_sdk/cli/__init__.py +++ b/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) diff --git a/singer_sdk/configuration/_dict_config.py b/singer_sdk/configuration/_dict_config.py index 021099adb..fd8217f01 100644 --- a/singer_sdk/configuration/_dict_config.py +++ b/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." diff --git a/singer_sdk/helpers/_util.py b/singer_sdk/helpers/_util.py index 0e1399043..d0079c40d 100644 --- a/singer_sdk/helpers/_util.py +++ b/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) diff --git a/singer_sdk/mapper_base.py b/singer_sdk/mapper_base.py index 977da61ea..b0be198bd 100644 --- a/singer_sdk/mapper_base.py +++ b/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 diff --git a/singer_sdk/plugin_base.py b/singer_sdk/plugin_base.py index 042ee5b54..57a14fe95 100644 --- a/singer_sdk/plugin_base.py +++ b/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() diff --git a/singer_sdk/tap_base.py b/singer_sdk/tap_base.py index fb6443a60..e6d999a02 100644 --- a/singer_sdk/tap_base.py +++ b/singer_sdk/tap_base.py @@ -12,7 +12,6 @@ import click from singer_sdk._singerlib import Catalog -from singer_sdk.cli import common_options from singer_sdk.configuration._dict_config import merge_missing_config_jsonschema from singer_sdk.exceptions import AbortedSyncFailedException, AbortedSyncPausedException from singer_sdk.helpers import _state @@ -42,7 +41,7 @@ class CliTestOptionValue(Enum): All = "all" Schema = "schema" - Disabled = False + Disabled = "disabled" class Tap(PluginBase, metaclass=abc.ABCMeta): @@ -459,108 +458,143 @@ def sync_all(self) -> None: # Command Line Execution - @classproperty - def cli(cls) -> t.Callable: # noqa: N805 - """Execute standard CLI handler for taps. + @classmethod + def invoke( # type: ignore[override] + cls: type[Tap], + *, + about: bool = False, + about_format: str | None = None, + config: tuple[str, ...] = (), + state: str | None = None, + catalog: str | None = None, + ) -> None: + """Invoke the tap's command line interface. - Returns: - A callable CLI object. + 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. + catalog: Use a Singer catalog file with the tap.", + state: Use a bookmarks file for incremental replication. """ + 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) - @common_options.PLUGIN_VERSION - @common_options.PLUGIN_ABOUT - @common_options.PLUGIN_ABOUT_FORMAT - @common_options.PLUGIN_CONFIG - @click.option( - "--discover", - is_flag=True, - help="Run the tap in discovery mode.", - ) - @click.option( - "--test", - is_flag=False, - flag_value=CliTestOptionValue.All.value, - default=CliTestOptionValue.Disabled, - help=( - "Use --test to sync a single record for each stream. " - "Use --test=schema to test schema output without syncing " - "records." - ), + tap = cls( + config=config_files, # type: ignore[arg-type] + state=state, + catalog=catalog, + parse_env_config=parse_env_config, + validate_config=True, ) - @click.option( - "--catalog", - help="Use a Singer catalog file with the tap.", - type=click.Path(), + tap.sync_all() + + @classmethod + def cb_discover( + cls: type[Tap], + ctx: click.Context, + param: click.Option, # noqa: ARG003 + value: bool, # noqa: FBT001 + ) -> None: + """CLI callback to run the tap in discovery mode. + + Args: + ctx: Click context. + param: Click option. + value: Whether to run in discovery mode. + """ + if not value: + return + + config_args = ctx.params.get("config", ()) + config_files, parse_env_config = cls.config_from_cli_args(*config_args) + tap = cls( + config=config_files, # type: ignore[arg-type] + parse_env_config=parse_env_config, + validate_config=False, ) - @click.option( - "--state", - help="Use a bookmarks file for incremental replication.", - type=click.Path(), + tap.run_discovery() + ctx.exit() + + @classmethod + def cb_test( + cls: type[Tap], + ctx: click.Context, + param: click.Option, # noqa: ARG003 + value: bool, # noqa: FBT001 + ) -> None: + """CLI callback to run the tap in test mode. + + Args: + ctx: Click context. + param: Click option. + value: Whether to run in test mode. + """ + if value == CliTestOptionValue.Disabled.value: + return + + config_args = ctx.params.get("config", ()) + config_files, parse_env_config = cls.config_from_cli_args(*config_args) + tap = cls( + config=config_files, # type: ignore[arg-type] + parse_env_config=parse_env_config, + validate_config=True, ) - @click.command( - help="Execute the Singer tap.", - context_settings={"help_option_names": ["--help"]}, + + if value == CliTestOptionValue.Schema.value: + tap.write_schemas() + else: + tap.run_connection_test() + + ctx.exit() + + @classmethod + def get_singer_command(cls: type[Tap]) -> click.Command: + """Execute standard CLI handler for taps. + + Returns: + A click.Command object. + """ + command = super().get_singer_command() + command.help = "Execute the Singer tap." + command.params.extend( + [ + click.Option( + ["--discover"], + is_flag=True, + help="Run the tap in discovery mode.", + callback=cls.cb_discover, + expose_value=False, + ), + click.Option( + ["--test"], + is_flag=False, + flag_value=CliTestOptionValue.All.value, + default=CliTestOptionValue.Disabled.value, + help=( + "Use --test to sync a single record for each stream. " + "Use --test=schema to test schema output without syncing " + "records." + ), + callback=cls.cb_test, + expose_value=False, + ), + click.Option( + ["--catalog"], + help="Use a Singer catalog file with the tap.", + type=click.Path(), + ), + click.Option( + ["--state"], + help="Use a bookmarks file for incremental replication.", + type=click.Path(), + ), + ], ) - def cli( - *, - version: bool = False, - about: bool = False, - discover: bool = False, - test: CliTestOptionValue = CliTestOptionValue.Disabled, - config: tuple[str, ...] = (), - state: str | None = None, - catalog: str | None = None, - about_format: str | None = None, - ) -> None: - """Handle command line execution. - - Args: - version: Display the package version. - about: Display package metadata and settings. - discover: Run the tap in discovery mode. - test: Test connectivity by syncing a single record and exiting. - about_format: Specify output style for `--about`. - config: Configuration file location or 'ENV' to use environment - variables. Accepts multiple inputs as a tuple. - catalog: Use a Singer catalog file with the tap.", - state: Use a bookmarks file for incremental replication. - """ - if version: - cls.print_version() - return - - if not about: - cls.print_version(print_fn=cls.logger.info) - else: - cls.print_about(output_format=about_format) - return - - validate_config: bool = True - if discover: - # Don't abort on validation failures - validate_config = False - - config_files, parse_env_config = cls.config_from_cli_args(*config) - tap = cls( # type: ignore[operator] - config=config_files or None, - state=state, - catalog=catalog, - parse_env_config=parse_env_config, - validate_config=validate_config, - ) - - if discover: - tap.run_discovery() - if test == CliTestOptionValue.All.value: - tap.run_connection_test() - elif test == CliTestOptionValue.All.value: - tap.run_connection_test() - elif test == CliTestOptionValue.Schema.value: - tap.write_schemas() - else: - tap.sync_all() - - return cli + + return command class SQLTap(Tap): diff --git a/singer_sdk/target_base.py b/singer_sdk/target_base.py index d44305d67..2fcb28314 100644 --- a/singer_sdk/target_base.py +++ b/singer_sdk/target_base.py @@ -12,7 +12,6 @@ import click from joblib import Parallel, delayed, parallel_backend -from singer_sdk.cli import common_options from singer_sdk.exceptions import RecordsWithoutSchemaException from singer_sdk.helpers._batch import BaseBatchFileEncoding from singer_sdk.helpers._classproperty import classproperty @@ -28,7 +27,6 @@ from singer_sdk.plugin_base import PluginBase if t.TYPE_CHECKING: - from io import FileIO from pathlib import PurePath from singer_sdk.sinks import Sink @@ -518,66 +516,55 @@ def _write_state_message(self, state: dict) -> None: # CLI handler - @classproperty - def cli(cls) -> t.Callable: # noqa: N805 - """Execute standard CLI handler for taps. + @classmethod + def invoke( # type: ignore[override] + cls: type[Target], + *, + about: bool = False, + about_format: str | None = None, + config: tuple[str, ...] = (), + file_input: t.IO[str] | None = None, + ) -> None: + """Invoke the target. - Returns: - A callable CLI object. + 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) - @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 target.", - context_settings={"help_option_names": ["--help"]}, + target = cls( + config=config_files, # type: ignore[arg-type] + validate_config=True, + parse_env_config=parse_env_config, ) - 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) - else: - cls.print_about(output_format=about_format) - return + target.listen(file_input) - validate_config: bool = True - - cls.print_version(print_fn=cls.logger.info) - - config_files, parse_env_config = cls.config_from_cli_args(*config) - target = cls( # type: ignore[operator] - config=config_files or None, - parse_env_config=parse_env_config, - validate_config=validate_config, - ) + @classmethod + def get_singer_command(cls: type[Target]) -> click.Command: + """Execute standard CLI handler for taps. - target.listen(file_input) + Returns: + A click.Command object. + """ + command = super().get_singer_command() + command.help = "Execute the Singer target." + command.params.extend( + [ + click.Option( + ["--input", "file_input"], + help="A path to read messages from instead of from standard in.", + type=click.File("r"), + ), + ], + ) - return cli + return command class SQLTarget(Target):