refactor: Consolidate config parsing for all plugin base classes

singer_sdk/mapper_base.py

@@ -9,7 +9,6 @@
 
 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
@@ -22,10 +21,6 @@
 class InlineMapper(PluginBase, SingerReader, metaclass=abc.ABCMeta):
     """Abstract base class for inline mappers."""
 
-    @classproperty
-    def _env_prefix(cls) -> str:  # noqa: N805
-        return f"{cls.name.upper().replace('-', '_')}_"
-
     @classproperty
     def capabilities(self) -> list[CapabilitiesEnum]:
         """Get capabilities.
@@ -159,15 +154,11 @@ def cli(
 
             cls.print_version(print_fn=cls.logger.info)
 
-            config_dict = merge_config_sources(
-                config,
-                cls.config_jsonschema,
-                cls._env_prefix,
-            )
-
+            config_files, parse_env_config = cls.config_from_cli_args(*config)
             mapper = cls(  # type: ignore[operator]
-                config=config_dict,
+                config=config_files or None,
                 validate_config=validate_config,
+                parse_env_config=parse_env_config,
             )
 
             if about:

singer_sdk/plugin_base.py

@@ -5,7 +5,7 @@
 import abc
 import logging
 import os
-from pathlib import PurePath
+from pathlib import Path, PurePath
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Callable, Mapping, cast
 
@@ -367,6 +367,40 @@ def print_about(
         formatter = about.AboutFormatter.get_formatter(output_format or "text")
         print(formatter.format_about(info))  # noqa: T201
 
+    @staticmethod
+    def config_from_cli_args(*args: str) -> tuple[list[Path], bool]:
+        """Parse CLI arguments into a config dictionary.
+
+        Args:
+            args: CLI arguments.
+
+        Raises:
+            FileNotFoundError: If the config file does not exist.
+
+        Returns:
+            A tuple containing the config dictionary and a boolean indicating whether
+            the config file was found.
+        """
+        config_files = []
+        parse_env_config = False
+
+        for config_path in args:
+            if config_path == "ENV":
+                # Allow parse from env vars:
+                parse_env_config = True
+                continue
+
+            # Validate config file paths before adding to list
+            if not Path(config_path).is_file():
+                raise FileNotFoundError(
+                    f"Could not locate config file at '{config_path}'."
+                    "Please check that the file exists.",
+                )
+
+            config_files.append(Path(config_path))
+
+        return config_files, parse_env_config
+
     @classproperty
     def cli(cls) -> Callable:  # noqa: N805
         """Handle command line execution.

singer_sdk/tap_base.py

@@ -7,7 +7,6 @@
 import contextlib
 import json
 from enum import Enum
-from pathlib import Path, PurePath
 from typing import TYPE_CHECKING, Any, Callable, Iterable, Sequence, cast
 
 import click
@@ -29,6 +28,8 @@
 from singer_sdk.plugin_base import PluginBase
 
 if TYPE_CHECKING:
+    from pathlib import PurePath
+
     from singer_sdk.streams import SQLStream, Stream
 
 STREAM_MAPS_CONFIG = "stream_maps"
@@ -495,9 +496,6 @@ def cli(
                     variables. Accepts multiple inputs as a tuple.
                 catalog: Use a Singer catalog file with the tap.",
                 state: Use a bookmarks file for incremental replication.
-
-            Raises:
-                FileNotFoundError: If the config file does not exist.
             """
             if version:
                 cls.print_version()
@@ -514,23 +512,7 @@ def cli(
                 # Don't abort on validation failures
                 validate_config = False
 
-            parse_env_config = False
-            config_files: list[PurePath] = []
-            for config_path in config:
-                if config_path == "ENV":
-                    # Allow parse from env vars:
-                    parse_env_config = True
-                    continue
-
-                # Validate config file paths before adding to list
-                if not Path(config_path).is_file():
-                    raise FileNotFoundError(
-                        f"Could not locate config file at '{config_path}'."
-                        "Please check that the file exists.",
-                    )
-
-                config_files.append(Path(config_path))
-
+            config_files, parse_env_config = cls.config_from_cli_args(*config)
             tap = cls(  # type: ignore[operator]
                 config=config_files or None,
                 state=state,

singer_sdk/target_base.py

@@ -7,7 +7,6 @@
 import json
 import sys
 import time
-from pathlib import Path, PurePath
 from typing import IO, TYPE_CHECKING, Callable, Counter
 
 import click
@@ -30,6 +29,7 @@
 
 if TYPE_CHECKING:
     from io import FileIO
+    from pathlib import PurePath
 
     from singer_sdk.sinks import Sink
 
@@ -551,9 +551,6 @@ def cli(
                     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.
-
-            Raises:
-                FileNotFoundError: If the config file does not exist.
             """
             if version:
                 cls.print_version()
@@ -569,23 +566,7 @@ def cli(
 
             cls.print_version(print_fn=cls.logger.info)
 
-            parse_env_config = False
-            config_files: list[PurePath] = []
-            for config_path in config:
-                if config_path == "ENV":
-                    # Allow parse from env vars:
-                    parse_env_config = True
-                    continue
-
-                # Validate config file paths before adding to list
-                if not Path(config_path).is_file():
-                    raise FileNotFoundError(
-                        f"Could not locate config file at '{config_path}'."
-                        "Please check that the file exists.",
-                    )
-
-                config_files.append(Path(config_path))
-
+            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,