diff --git a/singer_sdk/configuration/_dict_config.py b/singer_sdk/configuration/_dict_config.py index 23c89ca25..c75f016d3 100644 --- a/singer_sdk/configuration/_dict_config.py +++ b/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.", diff --git a/singer_sdk/helpers/_util.py b/singer_sdk/helpers/_util.py index c4c09383c..2ce3631dd 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: raise RuntimeError("Could not open file. Filepath not provided.") diff --git a/singer_sdk/mapper_base.py b/singer_sdk/mapper_base.py index 702963391..bb287da65 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.""" @@ -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 diff --git a/singer_sdk/plugin_base.py b/singer_sdk/plugin_base.py index d19be13e8..83d48ee7c 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 @@ -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, + ), + ], + ) diff --git a/singer_sdk/tap_base.py b/singer_sdk/tap_base.py index f2f24bd70..cee33396e 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.exceptions import AbortedSyncFailedException, AbortedSyncPausedException from singer_sdk.helpers import _state from singer_sdk.helpers._classproperty import classproperty @@ -40,7 +39,7 @@ class CliTestOptionValue(Enum): All = "all" Schema = "schema" - Disabled = False + Disabled = "disabled" class Tap(PluginBase, metaclass=abc.ABCMeta): @@ -431,108 +430,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_command(cls: type[Tap]) -> click.Command: + """Execute standard CLI handler for taps. + + Returns: + A click.Command object. + """ + command = super().get_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 1d8d97f4f..b46f85e63 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 @@ -516,66 +514,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_command(cls: type[Target]) -> click.Command: + """Execute standard CLI handler for taps. - target.listen(file_input) + Returns: + A click.Command object. + """ + command = super().get_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):